WeiCHEN
381b56dafc
Introduce internal/contracts/vocabulary for state/role/error sentinels, alias through docgov, and retarget hot production consumers so coordinator no longer imports docgov for vocabulary-only use.
316 lines
10 KiB
Go
316 lines
10 KiB
Go
package docflow
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/app/uow"
|
|
"chat-agent-system/internal/contracts/vocabulary"
|
|
"chat-agent-system/internal/docgov"
|
|
"chat-agent-system/internal/docrevision"
|
|
"chat-agent-system/internal/events"
|
|
)
|
|
|
|
type Action string
|
|
|
|
const (
|
|
ActionEdit Action = "document.edit"
|
|
ActionApprove Action = "document.approve"
|
|
ActionReject Action = "document.reject"
|
|
ActionCommit Action = "document.agent_commit"
|
|
)
|
|
|
|
type Principal struct {
|
|
UserID int64
|
|
Name string
|
|
Role vocabulary.UserRole
|
|
}
|
|
|
|
type HumanEditCommand struct {
|
|
DocID string
|
|
Request docrevision.HumanRevisionRequest
|
|
}
|
|
|
|
type AgentCommitCommand struct {
|
|
DocID string
|
|
CandidateID int64
|
|
Decision docrevision.CommitAgentDecision
|
|
}
|
|
|
|
type DecisionCommand struct {
|
|
DocID string
|
|
Comment string
|
|
RejectReason string
|
|
RollbackTo vocabulary.DocumentState
|
|
}
|
|
|
|
type Commands struct {
|
|
UoW uow.UnitOfWork
|
|
Gov *docgov.GovernanceEngine
|
|
Rev *docrevision.Service
|
|
Auth Authorizer
|
|
Events EventSink
|
|
Environment string
|
|
}
|
|
|
|
type EventSink interface {
|
|
Append(ctx context.Context, env events.Envelope, consumers []string) error
|
|
}
|
|
|
|
type Authorizer interface {
|
|
Authorize(ctx context.Context, p Principal, rec *docgov.DocumentRecord, action Action) error
|
|
}
|
|
|
|
type dutyAuthorizer struct{}
|
|
|
|
func DutyAuthorizer() Authorizer { return dutyAuthorizer{} }
|
|
|
|
func (dutyAuthorizer) Authorize(_ context.Context, p Principal, rec *docgov.DocumentRecord, action Action) error {
|
|
if rec == nil {
|
|
return uow.ErrPermissionDenied
|
|
}
|
|
switch action {
|
|
case ActionEdit, ActionCommit:
|
|
if rec.AuthorID != 0 && p.UserID != 0 && rec.AuthorID != p.UserID {
|
|
return uow.ErrPermissionDenied
|
|
}
|
|
case ActionApprove, ActionReject:
|
|
if rec.ApproverID != 0 && p.UserID != 0 && rec.ApproverID != p.UserID {
|
|
return uow.ErrPermissionDenied
|
|
}
|
|
default:
|
|
return uow.ErrPermissionDenied
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *Commands) run(ctx context.Context, docID string, p Principal, action Action, fn func(context.Context, uow.TxScope) error) error {
|
|
if c == nil {
|
|
return fmt.Errorf("docflow: commands required")
|
|
}
|
|
auth := c.authorizer()
|
|
if c.UoW == nil {
|
|
if err := c.authorizeUnlocked(ctx, docID, p, action); err != nil {
|
|
return err
|
|
}
|
|
return fn(ctx, nil)
|
|
}
|
|
return c.UoW.Within(ctx, func(scope uow.TxScope) error {
|
|
ctx = uow.WithScope(ctx, scope)
|
|
if docID != "" && c.Gov != nil && c.Gov.UsingPG() {
|
|
rec, err := (docgov.Repository{}).GetForUpdate(ctx, scope, docID)
|
|
if err != nil {
|
|
if action == ActionApprove || action == ActionReject {
|
|
return err
|
|
}
|
|
} else if err := auth.Authorize(ctx, p, rec, action); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return fn(ctx, scope)
|
|
})
|
|
}
|
|
|
|
func (c *Commands) authorizer() Authorizer {
|
|
if c != nil && c.Auth != nil {
|
|
return c.Auth
|
|
}
|
|
return DutyAuthorizer()
|
|
}
|
|
|
|
func (c *Commands) authorizeUnlocked(ctx context.Context, docID string, p Principal, action Action) error {
|
|
if docID == "" || c.Gov == nil {
|
|
return nil
|
|
}
|
|
rec, err := c.Gov.GetDocument(ctx, docID)
|
|
if err != nil || rec == nil {
|
|
if action == ActionApprove || action == ActionReject {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return uow.ErrPermissionDenied
|
|
}
|
|
return nil
|
|
}
|
|
return c.authorizer().Authorize(ctx, p, rec, action)
|
|
}
|
|
|
|
func (c *Commands) HumanEdit(ctx context.Context, p Principal, cmd HumanEditCommand) (*docrevision.HumanRevisionResult, error) {
|
|
var out *docrevision.HumanRevisionResult
|
|
err := c.run(ctx, cmd.DocID, p, ActionEdit, func(ctx context.Context, _ uow.TxScope) error {
|
|
res, err := c.Rev.CommitHumanRevision(ctx, cmd.Request)
|
|
out = res
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
func (c *Commands) AgentCommit(ctx context.Context, p Principal, cmd AgentCommitCommand) (*docrevision.RevisionState, error) {
|
|
var out *docrevision.RevisionState
|
|
err := c.run(ctx, cmd.DocID, p, ActionCommit, func(ctx context.Context, _ uow.TxScope) error {
|
|
st, err := c.Rev.CommitAgentVersion(ctx, cmd.DocID, cmd.CandidateID, cmd.Decision)
|
|
out = st
|
|
if err != nil || out == nil {
|
|
return err
|
|
}
|
|
rec, err := c.document(ctx, cmd.DocID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
body := []byte(fmt.Sprintf(`{"doc_id":%q,"version_no":%d}`, cmd.DocID, out.CurrentVersionNo))
|
|
if err := c.emit(ctx, rec, events.EventVersionCommitted,
|
|
fmt.Sprintf("document:%s:v%d", cmd.DocID, out.CurrentVersionNo),
|
|
fmt.Sprintf("commit:%s:%d", cmd.DocID, out.CurrentVersionNo),
|
|
"document", rec.DocID, int64(out.CurrentVersionNo), p, body); err != nil {
|
|
return err
|
|
}
|
|
return c.emitDownstream(ctx, rec, p, out, cmd.CandidateID)
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
func (c *Commands) Approve(ctx context.Context, p Principal, cmd DecisionCommand) (*docgov.DocumentRecord, error) {
|
|
var out *docgov.DocumentRecord
|
|
err := c.run(ctx, cmd.DocID, p, ActionApprove, func(ctx context.Context, _ uow.TxScope) error {
|
|
doc, err := c.Gov.ApproverApprove(ctx, cmd.DocID, p.UserID, p.Role, p.Name, cmd.Comment)
|
|
out = doc
|
|
if err != nil || out == nil {
|
|
return err
|
|
}
|
|
body := []byte(fmt.Sprintf(`{"doc_id":%q,"project_id":%d,"task_key":%q}`, out.DocID, out.ProjectID, out.TaskKey))
|
|
return c.emit(ctx, out, events.EventApproved,
|
|
fmt.Sprintf("document:%s:approved:%d", out.DocID, out.ID),
|
|
fmt.Sprintf("approved:%s:%d", out.DocID, out.ID),
|
|
"document_approval", out.DocID, int64(out.ID), p, body)
|
|
})
|
|
if err == nil && out != nil && c.UoW != nil && c.Gov != nil {
|
|
c.Gov.RunApprovedHooksAfterCommit(ctx, out)
|
|
}
|
|
return out, err
|
|
}
|
|
|
|
func (c *Commands) MaterializeAgent(ctx context.Context, p Principal, req docrevision.AgentMaterializeRequest) (*docrevision.AgentMaterializeResult, error) {
|
|
var out *docrevision.AgentMaterializeResult
|
|
err := c.run(ctx, req.DocID, p, ActionCommit, func(ctx context.Context, _ uow.TxScope) error {
|
|
res, err := c.Rev.MaterializeAgentOutput(ctx, req)
|
|
out = res
|
|
if err != nil || res == nil || res.Candidate == nil {
|
|
return err
|
|
}
|
|
rec, recErr := c.document(ctx, req.DocID)
|
|
if recErr != nil {
|
|
return recErr
|
|
}
|
|
if res.Candidate.Status == docrevision.CandidateNeedsConfirmation {
|
|
risk := ""
|
|
if res.Report != nil {
|
|
risk = res.Report.RiskLevel
|
|
}
|
|
body := []byte(fmt.Sprintf(`{"doc_id":%q,"candidate_id":%d,"risk":%q}`, req.DocID, res.Candidate.ID, risk))
|
|
return c.emit(ctx, rec, events.EventCandidateReady,
|
|
fmt.Sprintf("document:%s:candidate:%d", req.DocID, res.Candidate.ID),
|
|
fmt.Sprintf("candidate:%s:%d", req.DocID, res.Candidate.ID),
|
|
"document_candidate", rec.DocID, int64(res.Candidate.ID), p, body)
|
|
}
|
|
if res.Candidate.Status == docrevision.CandidateCommitted && res.State != nil {
|
|
body := []byte(fmt.Sprintf(`{"doc_id":%q,"version_no":%d}`, req.DocID, res.State.CurrentVersionNo))
|
|
if err := c.emit(ctx, rec, events.EventVersionCommitted,
|
|
fmt.Sprintf("document:%s:v%d", req.DocID, res.State.CurrentVersionNo),
|
|
fmt.Sprintf("commit:%s:%d", req.DocID, res.State.CurrentVersionNo),
|
|
"document", rec.DocID, int64(res.State.CurrentVersionNo), p, body); err != nil {
|
|
return err
|
|
}
|
|
return c.emitDownstream(ctx, rec, p, res.State, res.Candidate.ID)
|
|
}
|
|
return nil
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
// MaterializeAgentOutput satisfies bizdev.AgentOutputMaterializer so the squad
|
|
// path uses the unique Commands instance instead of calling docrevision directly.
|
|
func (c *Commands) MaterializeAgentOutput(ctx context.Context, req docrevision.AgentMaterializeRequest) (*docrevision.AgentMaterializeResult, error) {
|
|
return c.MaterializeAgent(ctx, Principal{UserID: req.AutoCommitUserID}, req)
|
|
}
|
|
|
|
func (c *Commands) Reject(ctx context.Context, p Principal, cmd DecisionCommand) (*docgov.DocumentRecord, error) {
|
|
var out *docgov.DocumentRecord
|
|
err := c.run(ctx, cmd.DocID, p, ActionReject, func(ctx context.Context, _ uow.TxScope) error {
|
|
doc, err := c.Gov.ApproverReject(ctx, cmd.DocID, p.UserID, p.Role, p.Name, cmd.RejectReason, cmd.RollbackTo)
|
|
out = doc
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
func (c *Commands) document(ctx context.Context, docID string) (*docgov.DocumentRecord, error) {
|
|
if c == nil || c.Gov == nil {
|
|
return nil, events.ErrInvalidScope
|
|
}
|
|
rec, err := c.Gov.GetDocument(ctx, docID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if rec == nil || rec.ProjectID <= 0 {
|
|
return nil, events.ErrInvalidScope
|
|
}
|
|
return rec, nil
|
|
}
|
|
|
|
func (c *Commands) emit(ctx context.Context, rec *docgov.DocumentRecord, eventName, eventID, idempotency, aggregateType, aggregateID string, version int64, p Principal, payload []byte) error {
|
|
if c == nil || c.Events == nil {
|
|
return nil
|
|
}
|
|
if rec == nil || rec.ProjectID <= 0 || strings.TrimSpace(c.Environment) == "" {
|
|
return events.ErrInvalidScope
|
|
}
|
|
if aggregateType == "" {
|
|
aggregateType = "document"
|
|
}
|
|
if aggregateID == "" {
|
|
aggregateID = rec.DocID
|
|
}
|
|
env := events.Envelope{
|
|
EventID: eventID, EventName: eventName, SchemaVersion: events.EnvelopeSchema,
|
|
Environment: c.Environment, ProjectID: rec.ProjectID,
|
|
AggregateType: aggregateType, AggregateID: aggregateID, AggregateVersion: version,
|
|
ActorRef: fmt.Sprintf("user:%d", p.UserID), IdempotencyKey: idempotency,
|
|
PayloadDigest: events.Digest(payload), Producer: "docflow",
|
|
OccurredAt: time.Now().UTC(), Payload: payload,
|
|
}
|
|
return c.Events.Append(ctx, env, events.LiveConsumers(eventName))
|
|
}
|
|
|
|
func (c *Commands) emitDownstream(ctx context.Context, rec *docgov.DocumentRecord, p Principal, st *docrevision.RevisionState, candidateID int64) error {
|
|
if c == nil || c.Rev == nil || st == nil || rec == nil || candidateID == 0 {
|
|
return nil
|
|
}
|
|
cand, err := c.Rev.GetCandidate(ctx, rec.DocID, candidateID)
|
|
if err != nil || cand == nil || cand.ChangeReportID == nil {
|
|
return nil
|
|
}
|
|
rep, err := c.Rev.GetChangeReport(ctx, rec.DocID, *cand.ChangeReportID)
|
|
if err != nil || rep == nil {
|
|
return nil
|
|
}
|
|
for _, imp := range rep.DocumentImpacts {
|
|
body := []byte(fmt.Sprintf(`{"doc_id":%q,"impacted_doc_id":%q,"version_no":%d}`, rec.DocID, imp.DocID, st.CurrentVersionNo))
|
|
if err := c.emit(ctx, rec, events.EventDownstreamImpact,
|
|
fmt.Sprintf("document:%s:impact:%s:v%d", rec.DocID, imp.DocID, st.CurrentVersionNo),
|
|
fmt.Sprintf("impact:%s:%s:%d", rec.DocID, imp.DocID, st.CurrentVersionNo),
|
|
"document_impact", rec.DocID+":"+imp.DocID, int64(st.CurrentVersionNo), p, body); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if rep.AnalysisStatus == docrevision.AnalysisUnassessed || rep.AnalysisStatus == docrevision.AnalysisFallback {
|
|
body := []byte(fmt.Sprintf(`{"doc_id":%q,"report_id":%d}`, rec.DocID, rep.ID))
|
|
return c.emit(ctx, rec, events.EventChangeReportDegraded,
|
|
fmt.Sprintf("document:%s:report:%d", rec.DocID, rep.ID),
|
|
fmt.Sprintf("degraded:%s:%d", rec.DocID, rep.ID),
|
|
"document_report", rec.DocID, int64(rep.ID), p, body)
|
|
}
|
|
return nil
|
|
}
|