Agent-System/internal/app/intentflow/projection.go

package intentflow

import (
	"context"
	"encoding/json"
	"strings"
	"sync"

	"chat-agent-system/internal/events"

	"github.com/jackc/pgx/v5"
)

// MemorySink records envelopes for tests. It still enforces live contracts.
type MemorySink struct {
	mu   sync.Mutex
	Envs []events.Envelope
}

func (m *MemorySink) Append(ctx context.Context, env events.Envelope, consumers []string) error {
	if err := events.AssertLiveConsumers(env.EventName, consumers); err != nil {
		return err
	}
	if err := events.ValidateEnvelope(env); err != nil {
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	m.Envs = append(m.Envs, env)
	return nil
}

func (m *MemorySink) All() []events.Envelope {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]events.Envelope, len(m.Envs))
	copy(out, m.Envs)
	return out
}

// Shadow is the only-forward projection of a plan. It is not SSOT.
type Shadow struct {
	PlanID      string
	Revision    int64
	Status      string
	StepID      string
	Environment string
	ProjectID   int64
	UserID      int64
	SessionID   string
	SummaryRef  string
}

// Projector applies intent.plan.* facts idempotently and only forward.
type Projector struct {
	mu     sync.Mutex
	events map[string]struct{}
	shadow map[string]Shadow
	cards  []CardProjection
	fail   bool
}

func NewProjector() *Projector {
	return &Projector{events: map[string]struct{}{}, shadow: map[string]Shadow{}}
}

func (p *Projector) SetFail(v bool) {
	if p == nil {
		return
	}
	p.mu.Lock()
	p.fail = v
	p.mu.Unlock()
}

func (p *Projector) ProjectCard(ctx context.Context, card CardProjection) error {
	if p == nil {
		return nil
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.fail {
		return events.ErrReplayApply
	}
	p.cards = append(p.cards, card)
	return nil
}

func (p *Projector) Cards() []CardProjection {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := make([]CardProjection, len(p.cards))
	copy(out, p.cards)
	return out
}

func (p *Projector) Shadow(planID string) (Shadow, bool) {
	p.mu.Lock()
	defer p.mu.Unlock()
	s, ok := p.shadow[planID]
	return s, ok
}

func (p *Projector) ApplyEnvelope(env events.Envelope) error {
	if p == nil {
		return nil
	}
	if err := events.ValidateEnvelope(env); err != nil {
		return err
	}
	if !strings.HasPrefix(env.EventName, "intent.plan.") {
		return nil
	}
	fact, err := decodePlanFact(env.Payload)
	if err != nil {
		return err
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if _, ok := p.events[env.EventID]; ok {
		return nil
	}
	p.events[env.EventID] = struct{}{}
	cur, ok := p.shadow[fact.PlanID]
	if ok && fact.Revision < cur.Revision {
		return nil
	}
	p.shadow[fact.PlanID] = Shadow{
		PlanID: fact.PlanID, Revision: fact.Revision, Status: fact.Status, StepID: fact.StepID,
		Environment: fact.Environment, ProjectID: fact.ProjectID, UserID: fact.UserID,
		SessionID: fact.SessionID, SummaryRef: fact.SummaryRef,
	}
	return nil
}

func (p *Projector) VisibleShadow(prin Principal, planID string) (Shadow, bool) {
	s, ok := p.Shadow(planID)
	if !ok {
		return Shadow{}, false
	}
	if s.Environment != prin.Environment || s.ProjectID != prin.ProjectID || s.SessionID != prin.SessionID || s.UserID != prin.UserID {
		return Shadow{}, false
	}
	return s, true
}

func decodePlanFact(body []byte) (planFact, error) {
	var fact planFact
	if err := json.Unmarshal(body, &fact); err != nil {
		return planFact{}, err
	}
	return fact, nil
}

// ProjectionConsumer applies intent.plan.* onto shadow tables inside the lease tx.
type ProjectionConsumer struct {
	Env   string
	Cards CardProjector
}

func (c *ProjectionConsumer) 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 !strings.HasPrefix(lease.EventName, "intent.plan.") {
		return nil
	}
	if err := events.ValidateEnvelope(lease.Envelope); err != nil {
		return events.ErrQuarantine
	}
	fact, err := decodePlanFact(lease.Envelope.Payload)
	if err != nil {
		return events.ErrQuarantine
	}
	if fact.UserID <= 0 || fact.SessionID == "" {
		return events.ErrQuarantine
	}
	tag, err := tx.Exec(ctx, `
INSERT INTO intent_plan_projections (
  source_event_id, plan_id, revision, status, step_id, environment, project_id, user_id, session_id, summary_ref
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (source_event_id) DO NOTHING`,
		lease.EventID, fact.PlanID, fact.Revision, fact.Status, fact.StepID,
		fact.Environment, fact.ProjectID, fact.UserID, fact.SessionID, fact.SummaryRef)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return nil
	}
	_, err = tx.Exec(ctx, `
INSERT INTO intent_plan_shadow (
  plan_id, revision, status, step_id, environment, project_id, user_id, session_id, summary_ref, updated_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9, now())
ON CONFLICT (plan_id) DO UPDATE
SET revision = EXCLUDED.revision,
    status = EXCLUDED.status,
    step_id = EXCLUDED.step_id,
    summary_ref = EXCLUDED.summary_ref,
    updated_at = now()
WHERE intent_plan_shadow.revision <= EXCLUDED.revision`,
		fact.PlanID, fact.Revision, fact.Status, fact.StepID,
		fact.Environment, fact.ProjectID, fact.UserID, fact.SessionID, fact.SummaryRef)
	if err != nil {
		return err
	}
	if c.Cards != nil {
		if err := c.Cards.ProjectCard(ctx, CardProjection{
			Principal: Principal{
				UserID: fact.UserID, Environment: fact.Environment,
				ProjectID: fact.ProjectID, SessionID: fact.SessionID,
			},
			PlanID: fact.PlanID, Revision: fact.Revision,
			Status: PlanStatus(fact.Status), SummaryRef: fact.SummaryRef,
		}); err != nil {
			return err
		}
	}
	return nil
}

// MuxProjection dispatches document vs intent-plan leases without changing claim semantics.
func MuxProjection(doc events.Handler, plan events.Handler) events.Handler {
	return func(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
		if strings.HasPrefix(lease.EventName, "intent.plan.") {
			if plan == nil {
				return nil
			}
			return plan(ctx, lease, tx)
		}
		if doc == nil {
			return nil
		}
		return doc(ctx, lease, tx)
	}
}