flex-auth/internal/emission/outbox.go

// Package emission delivers the decision-record emission guarantee declared in
// cadence.yaml (FLEX-WP-0031). A decision is released only after its event is
// durably committed to a local outbox (FLEX-DEC-2026-018); draining to
// audit-core is asynchronous.
package emission

import (
	"bytes"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/netkingdom/flex-auth/pkg/api"
)

// ClassPrefix prefixes every decision event class; the suffix is the effect.
const ClassPrefix = "flex-auth.decision."

// PlatformTenant is sent when a decision names no tenant, because audit-core
// requires one on every event.
const PlatformTenant = "tenant:platform"

const outboxFile = "outbox.jsonl"

// Class returns the event class for a decision effect.
func Class(effect api.DecisionEffect) string {
	return ClassPrefix + string(effect)
}

// Event is one audit-core.event envelope: exactly the fields audit-core's
// normalize() requires. Data carries the decision record itself.
type Event struct {
	ID            string `json:"id"`
	Type          string `json:"type"`
	Source        string `json:"source"`
	Subject       string `json:"subject"`
	Tenant        string `json:"tenant"`
	CorrelationID string `json:"correlation_id"`
	OccurredAt    string `json:"occurred_at"`
	Data          any    `json:"data"`
}

// entry is one committed outbox line.
type entry struct {
	Seq         uint64 `json:"seq"`
	CommittedAt string `json:"committed_at"`
	Event       Event  `json:"event"`
}

// Status is the local half of reconciliation: what this outbox committed, per
// class, and what it released without committing.
type Status struct {
	Source              string            `json:"source"`
	Committed           map[string]uint64 `json:"committed"`
	ReleasedUncommitted map[string]uint64 `json:"released_uncommitted"`
	// ReleasedUncommitted is held in memory: a restart during an outbox outage
	// loses it (FLEX-DEC-2026-018, consequences).
	Means string `json:"means"`
}

// Outbox is a durable, append-only, fsync-per-commit event log.
type Outbox struct {
	source string
	path   string
	clock  func() time.Time

	mu          sync.Mutex
	file        *os.File
	seq         uint64
	size        int64 // bytes committed; a failed commit is cut back to it
	committed   map[string]uint64
	uncommitted map[string]uint64
}

// Open opens or creates the outbox in dir and rebuilds committed counts from it.
func Open(dir, source string) (*Outbox, error) {
	if source == "" {
		return nil, errors.New("emission source is required")
	}
	if err := os.MkdirAll(dir, 0o750); err != nil {
		return nil, fmt.Errorf("create outbox directory: %w", err)
	}
	o := &Outbox{
		source:      source,
		path:        filepath.Join(dir, outboxFile),
		clock:       time.Now,
		committed:   map[string]uint64{},
		uncommitted: map[string]uint64{},
	}
	if err := o.load(); err != nil {
		return nil, err
	}
	file, err := os.OpenFile(o.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640)
	if err != nil {
		return nil, fmt.Errorf("open outbox: %w", err)
	}
	o.file = file
	return o, nil
}

func (o *Outbox) load() error {
	data, err := os.ReadFile(o.path)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("read outbox: %w", err)
	}
	good := 0
	for line := 1; good < len(data); line++ {
		end := bytes.IndexByte(data[good:], '\n')
		if end < 0 {
			// A torn final line is a write whose sync never returned, so its
			// decision was never released as committed. Drop it so the next
			// append does not bury it mid-file.
			o.size = int64(good)
			return os.Truncate(o.path, o.size)
		}
		var e entry
		if err := json.Unmarshal(data[good:good+end], &e); err != nil {
			return fmt.Errorf("outbox line %d is corrupt: %w", line, err)
		}
		o.seq = e.Seq
		o.committed[e.Event.Type]++
		good += end + 1
	}
	o.size = int64(good)
	return nil
}

// SetClock overrides the commit clock. Tests use it.
func (o *Outbox) SetClock(clock func() time.Time) {
	o.mu.Lock()
	defer o.mu.Unlock()
	o.clock = clock
}

// Close closes the outbox file.
func (o *Outbox) Close() error {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.file.Close()
}

// Append commits one decision; it satisfies decision.DecisionRecorder.
func (o *Outbox) Append(decision api.DecisionEnvelope) error {
	return o.AppendBatch([]api.DecisionEnvelope{decision})
}

// AppendBatch commits decisions with one sync, so a batch check pays the
// durable-write cost once rather than per resource.
func (o *Outbox) AppendBatch(decisions []api.DecisionEnvelope) error {
	o.mu.Lock()
	defer o.mu.Unlock()
	now := o.clock().UTC().Format(time.RFC3339Nano)
	var buf []byte
	seq := o.seq
	for _, decision := range decisions {
		id, err := eventID()
		if err != nil {
			return err
		}
		seq++
		line, err := json.Marshal(entry{Seq: seq, CommittedAt: now, Event: o.event(id, now, decision)})
		if err != nil {
			return fmt.Errorf("marshal outbox entry: %w", err)
		}
		buf = append(append(buf, line...), '\n')
	}
	if _, err := o.file.Write(buf); err != nil {
		return o.rollback(fmt.Errorf("write outbox: %w", err))
	}
	if err := o.file.Sync(); err != nil {
		return o.rollback(fmt.Errorf("sync outbox: %w", err))
	}
	o.seq = seq
	o.size += int64(len(buf))
	for _, decision := range decisions {
		o.committed[Class(decision.Effect)]++
	}
	return nil
}

// rollback cuts a failed commit back to the last committed size, so a partial
// line never sits in front of the next commit. The commit error is returned
// either way; a failed cut is reported alongside it.
func (o *Outbox) rollback(cause error) error {
	if err := o.file.Truncate(o.size); err != nil {
		return errors.Join(cause, fmt.Errorf("truncate outbox after failed commit: %w", err))
	}
	return cause
}

// NoteReleasedUncommitted counts a restriction released after its commit
// failed. Every non-zero count is a finding.
func (o *Outbox) NoteReleasedUncommitted(effect api.DecisionEffect) {
	o.mu.Lock()
	defer o.mu.Unlock()
	o.uncommitted[Class(effect)]++
}

// Status reports committed and released-uncommitted counts per class.
func (o *Outbox) Status() Status {
	o.mu.Lock()
	defer o.mu.Unlock()
	status := Status{
		Source:              o.source,
		Committed:           map[string]uint64{},
		ReleasedUncommitted: map[string]uint64{},
		Means: "committed counts this outbox holds since it was created; " +
			"released_uncommitted is decisions released without a record " +
			"since this process started, and any non-zero value is a finding",
	}
	for _, effect := range api.DecisionEffects() {
		class := Class(effect)
		status.Committed[class] = o.committed[class]
		status.ReleasedUncommitted[class] = o.uncommitted[class]
	}
	return status
}

func (o *Outbox) event(id, at string, decision api.DecisionEnvelope) Event {
	tenant := decision.Resource.Tenant
	if decision.Binding != nil && decision.Binding.Tenant != "" {
		tenant = decision.Binding.Tenant
	}
	if tenant == "" {
		tenant = PlatformTenant
	}
	return Event{
		ID:            id,
		Type:          Class(decision.Effect),
		Source:        o.source,
		Subject:       resourceSubject(decision.Resource),
		Tenant:        tenant,
		CorrelationID: decision.ID,
		OccurredAt:    at,
		Data:          decision,
	}
}

func resourceSubject(resource api.ResourceRef) string {
	subject := resource.ID
	if resource.Type != "" {
		subject = resource.Type + "/" + subject
	}
	if resource.System != "" {
		subject = resource.System + "/" + subject
	}
	return subject
}

// eventID is random, not derived from the decision: decision ids are content
// digests, so two identical checks share one, and audit-core's idempotency key
// would merge two real decisions into one.
func eventID() (string, error) {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		return "", fmt.Errorf("event id: %w", err)
	}
	return "flex-auth-evt:" + hex.EncodeToString(b[:]), nil
}