97 lines
2.8 KiB
Go
97 lines
2.8 KiB
Go
package docproject
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
|
|
"chat-agent-system/internal/docgov"
|
|
"chat-agent-system/internal/events"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
)
|
|
|
|
// EventConsumer applies document.agent_version_committed onto ledger head pointers.
|
|
// Closures/producer already project in the approve/commit UoW; this path is idempotent no-replace.
|
|
type EventConsumer struct {
|
|
Docs DocumentLookup
|
|
Env string
|
|
}
|
|
|
|
type DocumentLookup interface {
|
|
GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error)
|
|
}
|
|
|
|
func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
|
|
if c == nil {
|
|
return events.ErrUnknownConsumer
|
|
}
|
|
principal := events.ServicePrincipal{
|
|
ID: "projection-" + c.Env, Environment: c.Env, Consumer: events.ConsumerProjection,
|
|
Actions: []string{"projection.apply"},
|
|
}
|
|
if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "projection.apply"); err != nil {
|
|
return events.ErrQuarantine
|
|
}
|
|
if lease.EventName != events.EventVersionCommitted {
|
|
return nil
|
|
}
|
|
docID := lease.Envelope.AggregateID
|
|
versionNo := int(lease.Envelope.AggregateVersion)
|
|
var payload map[string]any
|
|
_ = json.Unmarshal(lease.Envelope.Payload, &payload)
|
|
if payload != nil {
|
|
if id, ok := payload["doc_id"].(string); ok && id != "" {
|
|
docID = id
|
|
}
|
|
if v, ok := payload["version_no"].(float64); ok {
|
|
versionNo = int(v)
|
|
}
|
|
}
|
|
if c.Docs == nil || docID == "" {
|
|
return events.ErrQuarantine
|
|
}
|
|
doc, err := c.Docs.GetDocument(ctx, docID)
|
|
if err != nil || doc == nil {
|
|
return events.ErrQuarantine
|
|
}
|
|
if doc.ProjectID != lease.ProjectID {
|
|
return events.ErrQuarantine
|
|
}
|
|
var versionID, snapshotID int64
|
|
var sha string
|
|
err = tx.QueryRow(ctx, `
|
|
SELECT v.id, COALESCE(v.head_snapshot_id,0), COALESCE(s.content_sha256,'')
|
|
FROM document_versions v
|
|
LEFT JOIN document_snapshots s ON s.id = v.head_snapshot_id
|
|
WHERE v.doc_id=$1 AND v.version_no=$2
|
|
`, docID, versionNo).Scan(&versionID, &snapshotID, &sha)
|
|
if err != nil {
|
|
return events.ErrQuarantine
|
|
}
|
|
tag, err := tx.Exec(ctx, `
|
|
INSERT INTO event_projection_outcomes (source_event_id, doc_id, version_no, snapshot_id, applied)
|
|
VALUES ($1,$2,$3,$4,false)
|
|
ON CONFLICT (source_event_id) DO NOTHING
|
|
`, lease.EventID, docID, versionNo, snapshotID)
|
|
if err != nil {
|
|
var pgErr *pgconn.PgError
|
|
if errors.As(err, &pgErr) && pgErr.Code == "23505" {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return nil
|
|
}
|
|
applied, err := (docgov.Repository{}).ProjectHeadNoReplace(ctx, tx, docID, versionID, snapshotID, versionNo, sha)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.Exec(ctx, `
|
|
UPDATE event_projection_outcomes SET applied=$2, snapshot_id=$3 WHERE source_event_id=$1
|
|
`, lease.EventID, applied, snapshotID)
|
|
return err
|
|
}
|