fluid-core/cmd/fluid/commands.go

package main

import (
	"context"
	"crypto/ed25519"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"sort"
	"strings"
	"text/tabwriter"
	"time"

	"gopkg.in/yaml.v3"
	_ "modernc.org/sqlite"

	"github.com/tegwick/fluid-core/internal/contract"
	"github.com/tegwick/fluid-core/internal/evidence"
	"github.com/tegwick/fluid-core/internal/intent"
	"github.com/tegwick/fluid-core/internal/policy"
	"github.com/tegwick/fluid-core/internal/publish"
	"github.com/tegwick/fluid-core/internal/signing"
)

// open connects to the evidence store for this invocation.
func (g globals) open(ctx context.Context) (*evidence.SQLStore, error) {
	return evidence.OpenSQLite(ctx, g.storePath())
}

func out() *tabwriter.Writer {
	return tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0)
}

// ---------- intent ----------

func runIntent(ctx context.Context, g globals, args []string) error {
	if len(args) == 0 {
		return errors.New("intent needs a subcommand: put, activate, show")
	}
	iface, err := g.requireInterface()
	if err != nil {
		return err
	}
	store, err := g.open(ctx)
	if err != nil {
		return err
	}
	defer store.Close()

	is := intent.New(store, contract.InterfaceID(iface))

	switch args[0] {
	case "put":
		fs := newFlagSet("intent put")
		version := fs.String("version", "", "intent version label, such as IEI-1")
		file := fs.String("file", "", "path to the InterfaceEvolutionIntent document")
		activate := fs.Bool("activate", false, "make this version the governing intent")
		if err := fs.Parse(args[1:]); err != nil {
			return err
		}
		if *version == "" || *file == "" {
			return errors.New("intent put needs --version and --file")
		}

		doc, err := os.ReadFile(*file)
		if err != nil {
			return err
		}
		v, err := is.Put(ctx, *version, string(doc))
		if err != nil {
			return err
		}
		fmt.Printf("recorded %s\n  digest %s\n  mode   %s\n", v.Version, v.Digest, v.Mode)

		if *activate {
			if err := is.SetActive(ctx, v.Version); err != nil {
				return err
			}
			fmt.Printf("  active yes\n")
		}
		return nil

	case "activate":
		if len(args) < 2 {
			return errors.New("intent activate needs a version")
		}
		if err := is.SetActive(ctx, args[1]); err != nil {
			return err
		}
		fmt.Printf("%s is now the governing intent\n", args[1])
		return nil

	case "show":
		var v intent.Version
		if len(args) > 1 {
			v, err = is.Get(ctx, args[1])
		} else {
			v, err = is.Active(ctx)
		}
		if err != nil {
			return err
		}
		fmt.Printf("version   %s\ndigest    %s\nmode      %s\nrecorded  %s\n\n",
			v.Version, v.Digest, v.Mode, v.RecordedAt.Format(time.RFC3339))
		fmt.Println(v.Document)
		return nil

	default:
		return fmt.Errorf("unknown intent subcommand %q", args[0])
	}
}

// ---------- revision ----------

func runRevision(ctx context.Context, g globals, args []string) error {
	if len(args) == 0 {
		return errors.New("revision needs a subcommand: publish, list, show")
	}
	iface, err := g.requireInterface()
	if err != nil {
		return err
	}
	store, err := g.open(ctx)
	if err != nil {
		return err
	}
	defer store.Close()

	switch args[0] {
	case "publish":
		fs := newFlagSet("revision publish")
		file := fs.String("file", "", "path to the revision descriptor (YAML or JSON)")
		keyFile := fs.String("key-file", os.Getenv("FLUID_SIGNING_KEY"), "base64 ed25519 signing key file")
		keyID := fs.String("key-id", envOr("FLUID_SIGNING_KEY_ID", "dev"), "signing key identifier")
		ephemeral := fs.Bool("ephemeral-key", false, "sign with a throwaway key (development only)")
		classes := fs.String("adaptation-classes", "", "comma-separated adaptation classes")
		complexity := fs.Float64("complexity-delta", 0, "measured complexity impact")
		share := fs.Float64("traffic-share", 0, "requested traffic share")
		approvedBy := fs.String("approved-by", "", "authorizing operator; required by the default gate")
		if err := fs.Parse(args[1:]); err != nil {
			return err
		}
		if *file == "" {
			return errors.New("revision publish needs --file")
		}

		raw, err := os.ReadFile(*file)
		if err != nil {
			return err
		}
		var doc contract.RevisionDescriptorDocument
		if err := yaml.Unmarshal(raw, &doc); err != nil {
			return fmt.Errorf("parse descriptor: %w", err)
		}
		d := doc.Revision

		if string(d.Interface) != iface {
			return fmt.Errorf("descriptor is for interface %q, but --interface is %q", d.Interface, iface)
		}
		if d.ID == "" {
			return errors.New("descriptor has no revision id")
		}

		// Publishing goes through the same pipeline the control plane uses.
		// A CLI that could write a revision straight into the store would be a
		// way around the deterministic policy gate, which would make the gate
		// decorative for anyone with shell access.
		signer, err := loadSigner(*keyID, *keyFile, *ephemeral)
		if err != nil {
			return err
		}

		intents := intent.New(store, contract.InterfaceID(iface))
		pipeline, err := publish.New(publish.Options{
			Gate:    policy.NewGate(policy.DefaultLimits()),
			Signer:  signer,
			Store:   store,
			Intents: intents,
		})
		if err != nil {
			return err
		}

		var approver *contract.Actor
		if *approvedBy != "" {
			approver = &contract.Actor{Type: contract.ActorTypeHuman, ID: *approvedBy}
		}

		candidate := publish.NewCandidate(d, contract.Actor{
			Type: contract.ActorTypeHuman, ID: operator(),
		})

		verified, report, err := pipeline.Run(ctx, candidate, publish.PromotionRequest{
			AdaptationClasses:     parseClasses(*classes),
			ComplexityDelta:       *complexity,
			RequestedTrafficShare: *share,
			Approved:              approver != nil,
			ApprovedBy:            approver,
		})
		if err != nil {
			printReport(report)
			return err
		}
		if err := pipeline.Publish(ctx, verified); err != nil {
			return err
		}

		printReport(report)
		fmt.Printf("\npublished %s (%s), governed by %s, signed by %s\n",
			d.ID, d.State, d.Intent.Version, verified.Descriptor().Signature.KeyID)
		return nil

	case "list":
		records, err := store.Records(ctx, contract.KindRevision)
		if err != nil {
			return err
		}
		if len(records) == 0 {
			fmt.Println("no revisions published")
			return nil
		}

		ids := make([]string, 0, len(records))
		for id := range records {
			ids = append(ids, id)
		}
		sort.Strings(ids)

		w := out()
		fmt.Fprintln(w, "REVISION\tSTATE\tINTENT\tCOMPATIBILITY\tUPSTREAM")
		for _, id := range ids {
			var d contract.Revision
			if err := json.Unmarshal(records[id], &d); err != nil {
				fmt.Fprintf(w, "%s\t(unreadable)\t\t\t\n", id)
				continue
			}
			fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
				d.ID, d.State, d.Intent.Version, d.Policy.Compatibility, d.Runtime.Upstream)
		}
		return w.Flush()

	case "show":
		if len(args) < 2 {
			return errors.New("revision show needs a revision id")
		}
		body, err := store.Record(ctx, contract.KindRevision, args[1])
		if err != nil {
			return err
		}
		var pretty any
		_ = json.Unmarshal(body, &pretty)
		enc := json.NewEncoder(os.Stdout)
		enc.SetIndent("", "  ")
		return enc.Encode(pretty)

	default:
		return fmt.Errorf("unknown revision subcommand %q", args[0])
	}
}

// ---------- policy ----------

const policyRecordID = "__routing_policy__"

func runPolicy(ctx context.Context, g globals, args []string) error {
	if len(args) == 0 {
		return errors.New("policy needs a subcommand: put, show")
	}
	iface, err := g.requireInterface()
	if err != nil {
		return err
	}
	store, err := g.open(ctx)
	if err != nil {
		return err
	}
	defer store.Close()

	switch args[0] {
	case "put":
		fs := newFlagSet("policy put")
		file := fs.String("file", "", "path to the routing policy")
		if err := fs.Parse(args[1:]); err != nil {
			return err
		}
		if *file == "" {
			return errors.New("policy put needs --file")
		}

		raw, err := os.ReadFile(*file)
		if err != nil {
			return err
		}
		var doc contract.RoutingPolicyDocument
		if err := yaml.Unmarshal(raw, &doc); err != nil {
			return fmt.Errorf("parse policy: %w", err)
		}
		p := doc.RoutingPolicy

		if string(p.Interface) != iface {
			return fmt.Errorf("policy is for interface %q, but --interface is %q", p.Interface, iface)
		}

		// Generations are monotonic in the registry; catching a regression here
		// avoids shipping a policy the gateway will silently refuse.
		if existing, err := store.Record(ctx, contract.KindRoutingPolicy, policyRecordID); err == nil {
			var prev contract.RoutingPolicy
			if json.Unmarshal(existing, &prev) == nil && p.Generation <= prev.Generation {
				return fmt.Errorf("policy generation %d is not newer than the installed %d",
					p.Generation, prev.Generation)
			}
		}

		body, err := json.Marshal(p)
		if err != nil {
			return err
		}
		if err := store.PutRecord(ctx, contract.KindRoutingPolicy, policyRecordID, body); err != nil {
			return err
		}

		if err := store.AppendEvent(ctx, contract.FluidEvent{
			SchemaVersion: "0.1",
			ID:            contract.EventID(fmt.Sprintf("EV-policy-%d", time.Now().UnixNano())),
			OccurredAt:    time.Now().UTC(),
			EntityType:    contract.FluidEventEntityTypeRoutingPolicy,
			EntityID:      fmt.Sprintf("generation-%d", p.Generation),
			EventType:     "ROUTING_POLICY_INSTALLED",
			Actor:         contract.Actor{Type: contract.ActorTypeHuman, ID: operator()},
			Reason:        fmt.Sprintf("default %s, %d rule(s)", p.DefaultRevision, len(p.Rules)),
		}); err != nil {
			return err
		}

		fmt.Printf("installed routing policy generation %d (default %s, %d rules)\n",
			p.Generation, p.DefaultRevision, len(p.Rules))
		return nil

	case "show":
		body, err := store.Record(ctx, contract.KindRoutingPolicy, policyRecordID)
		if err != nil {
			if errors.Is(err, evidence.ErrNotFound) {
				fmt.Println("no routing policy installed")
				return nil
			}
			return err
		}
		var pretty any
		_ = json.Unmarshal(body, &pretty)
		enc := json.NewEncoder(os.Stdout)
		enc.SetIndent("", "  ")
		return enc.Encode(pretty)

	default:
		return fmt.Errorf("unknown policy subcommand %q", args[0])
	}
}

// ---------- events and telemetry ----------

func runEvents(ctx context.Context, g globals, args []string) error {
	fs := newFlagSet("events")
	entity := fs.String("entity", "", "filter by entity id, such as R-1")
	kind := fs.String("type", "", "filter by event type")
	limit := fs.Int("limit", 50, "maximum events to show")
	if err := fs.Parse(args); err != nil {
		return err
	}

	store, err := g.open(ctx)
	if err != nil {
		return err
	}
	defer store.Close()

	events, err := store.Events(ctx, evidence.EventFilter{
		EntityID:  *entity,
		EventType: *kind,
		Limit:     *limit,
	})
	if err != nil {
		return err
	}
	if len(events) == 0 {
		fmt.Println("no events")
		return nil
	}

	w := out()
	fmt.Fprintln(w, "WHEN\tENTITY\tTYPE\tACTOR\tREASON")
	for _, ev := range events {
		fmt.Fprintf(w, "%s\t%s %s\t%s\t%s:%s\t%s\n",
			ev.OccurredAt.Format(time.RFC3339), ev.EntityType, ev.EntityID,
			ev.EventType, ev.Actor.Type, ev.Actor.ID, oneLine(ev.Reason))
	}
	return w.Flush()
}

func runTelemetry(ctx context.Context, g globals, args []string) error {
	fs := newFlagSet("telemetry")
	revision := fs.String("revision", "", "filter by revision")
	kind := fs.String("kind", "", "filter by event kind")
	limit := fs.Int("limit", 50, "maximum rows to show")
	if err := fs.Parse(args); err != nil {
		return err
	}

	store, err := g.open(ctx)
	if err != nil {
		return err
	}
	defer store.Close()

	rows, err := store.Telemetry(ctx, evidence.TelemetryFilter{
		InterfaceID: contract.InterfaceID(g.iface),
		Revision:    contract.RevisionID(*revision),
		Kind:        contract.FluidTelemetryKind(*kind),
		Limit:       *limit,
	})
	if err != nil {
		return err
	}
	if len(rows) == 0 {
		fmt.Println("no telemetry")
		return nil
	}

	w := out()
	fmt.Fprintln(w, "WHEN\tKIND\tREVISION\tRESOLVED BY\tROUTE\tSTATUS")
	for _, ev := range rows {
		route, status := "", ""
		if ev.Request != nil {
			route = ev.Request.Route
			if ev.Request.Status != nil {
				status = fmt.Sprint(*ev.Request.Status)
			}
		}
		reason := ""
		if ev.Resolution != nil {
			reason = string(ev.Resolution.Reason)
		}
		fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\n",
			ev.OccurredAt.Format(time.RFC3339), ev.Kind, deref(ev.Revision), reason, route, status)
	}
	return w.Flush()
}

// ---------- audit ----------

// runAudit reconstructs the chain behind a revision.
//
// This is the command that has to answer the eleven questions in
// FluidAPIStandards.md section 25. It is deliberately built from events rather
// than from summary records: the records say what is true now, the events say
// how it came to be.
func runAudit(ctx context.Context, g globals, args []string) error {
	if len(args) < 2 || args[0] != "trace" {
		return errors.New("usage: fluid audit trace <revision>")
	}
	target := args[1]

	iface, err := g.requireInterface()
	if err != nil {
		return err
	}
	store, err := g.open(ctx)
	if err != nil {
		return err
	}
	defer store.Close()

	events, err := store.Events(ctx, evidence.EventFilter{EntityID: target})
	if err != nil {
		return err
	}
	if len(events) == 0 {
		return fmt.Errorf("no history recorded for %s", target)
	}

	fmt.Printf("Audit trace for %s\n\n", target)

	is := intent.New(store, contract.InterfaceID(iface))
	if v, err := is.GoverningIntent(ctx, contract.RevisionID(target)); err == nil {
		fmt.Printf("Governed by %s (%s), authority mode %s\n\n", v.Version, v.Digest, v.Mode)
	} else {
		fmt.Printf("Governed by: UNKNOWN — no intent binding recorded\n\n")
	}

	w := out()
	fmt.Fprintln(w, "WHEN\tEVENT\tACTOR\tINPUTS\tREASON")
	for _, ev := range events {
		fmt.Fprintf(w, "%s\t%s\t%s:%s\t%s\t%s\n",
			ev.OccurredAt.Format(time.RFC3339), ev.EventType,
			ev.Actor.Type, ev.Actor.ID,
			strings.Join(ev.Inputs, ","), oneLine(ev.Reason))
	}
	if err := w.Flush(); err != nil {
		return err
	}

	count, err := store.Telemetry(ctx, evidence.TelemetryFilter{Revision: contract.RevisionID(target)})
	if err == nil {
		fmt.Printf("\n%d telemetry event(s) recorded against %s\n", len(count), target)
	}
	return nil
}

// ---------- helpers ----------

func newFlagSet(name string) *flagSet { return &flagSet{fs: newStdFlagSet(name)} }

func oneLine(s string) string {
	return strings.Join(strings.Fields(s), " ")
}

func deref(s *contract.RevisionID) string {
	if s == nil {
		return ""
	}
	return string(*s)
}

func operator() string {
	if v := os.Getenv("FLUID_OPERATOR"); v != "" {
		return v
	}
	if v := os.Getenv("USER"); v != "" {
		return v
	}
	return "unknown"
}

// parseClasses splits a comma-separated adaptation class list.
func parseClasses(s string) []contract.AdaptationClass {
	if strings.TrimSpace(s) == "" {
		return nil
	}
	var out []contract.AdaptationClass
	for _, part := range strings.Split(s, ",") {
		if p := strings.TrimSpace(part); p != "" {
			out = append(out, contract.AdaptationClass(p))
		}
	}
	return out
}

// printReport renders the pipeline outcome stage by stage, so a rejection says
// which gate refused and why rather than only that it failed.
func printReport(r publish.Report) {
	if len(r.Stages) == 0 {
		return
	}
	w := out()
	fmt.Fprintln(w, "STAGE\tRESULT\tDETAIL")
	for _, s := range r.Stages {
		verdict := "pass"
		if !s.Passed {
			verdict = "FAIL"
		}
		fmt.Fprintf(w, "%s\t%s\t%s\n", s.Stage, verdict, oneLine(s.Detail))
	}
	_ = w.Flush()
}

// loadSigner resolves the signing key for a publish.
//
// An ephemeral key must be requested explicitly: signing with a key nobody
// trusts produces revisions the router will refuse, and finding that out at
// publish time is far better than at traffic time.
func loadSigner(keyID, keyFile string, ephemeral bool) (*signing.Signer, error) {
	if keyFile != "" {
		raw, err := os.ReadFile(keyFile)
		if err != nil {
			return nil, fmt.Errorf("read signing key: %w", err)
		}
		decoded, err := base64.StdEncoding.DecodeString(strings.TrimSpace(string(raw)))
		if err != nil {
			return nil, fmt.Errorf("signing key is not valid base64: %w", err)
		}
		return signing.NewSigner(keyID, ed25519.PrivateKey(decoded))
	}
	if ephemeral {
		signer, _, err := signing.GenerateKey(keyID)
		if err != nil {
			return nil, err
		}
		fmt.Fprintln(os.Stderr,
			"warning: signing with an ephemeral key; this revision will not verify after the key is gone")
		return signer, nil
	}
	return nil, errors.New("no signing key: pass --key-file, or --ephemeral-key for development")
}