mesh-catalog/modules/distribution/cmd/distribution-tools/main.go

// Command distribution-tools serves distribution's seat verbs and tools (novox/hq design 51): the verbs of
// container-image-registry and oci-artifact-registry, each as `<seat>.<verb>`, and registry_incoming, registry_holds,
// registry_old_name and registry_direct_writes. The runtime launches it on distribution's own account; the caller of
// every verb is the one the tool runner proved (stdio.CallerOf), never a name in the arguments. stdout is the MCP
// channel; what it says, it says on stderr.
package main

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/health"
	"git.novox.be/novox/mesh-sdk/go/parts"

	"distribution/internal/holder"
)

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, "[distribution-tools] "+format+"\n", a...) }

// kept is one state of distribution's, through the runtime.
type kept[T any] struct {
	state string
	key   func(T) string
}

func (k kept[T]) Get(key string) (T, bool, error) {
	var v T
	e, err := stdio.State(k.state).Get(key)
	if err != nil || e == nil {
		return v, false, err
	}
	return v, true, json.Unmarshal(e.Value, &v)
}
func (k kept[T]) Put(v T) error {
	_, err := stdio.State(k.state).Put(k.key(v), v)
	return err
}
func (k kept[T]) Delete(key string) error { return stdio.State(k.state).Delete(key) }
func (k kept[T]) List() ([]T, error) {
	keys, err := stdio.State(k.state).Keys()
	if err != nil {
		return nil, err
	}
	var out []T
	for _, key := range keys {
		if v, ok, err := k.Get(key); err == nil && ok {
			out = append(out, v)
		}
	}
	return out, nil
}

func say(seat, event string, body holder.Event) error {
	return stdio.SeatPublish("mesh.seat."+seat+".event."+event, eventBody(event, body), body.ID)
}

func finding(kind, subject, words string, raised bool) {
	key := health.FindingKey(health.ScopeSeat, "distribution", kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject))
	if !raised {
		_ = stdio.State(health.FindingsState).Delete(key)
		return
	}
	f := health.Finding{V: health.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:],
		Subject: health.Subject{Scope: health.ScopeSeat, ID: "distribution"}, Severity: health.Warning,
		Summary: words, Headline: words, Explanation: words, Resolved: "cleared once it no longer holds",
		Source: kind, Observer: "distribution", At: time.Now().UTC()}
	if _, err := stdio.State(health.FindingsState).Put(key, f); err != nil {
		logf("the finding %s could not be written: %v", key, err)
	}
}

// closeBootstrap asks the program to discard the bootstrap's one-off credential once a publish is answered (design 51
// *On a fresh mesh*): the window bootstrap-push closed in the state windows, which the program follows and then removes
// the file. The tools bundle runs in the tool runner's account, which cannot remove a file in distribution's state
// directory. Once per process; the window's record says it to the next.
var closeBootstrap func()

func bootstrapCloser(windows kept[holder.Window]) func() {
	return sync.OnceFunc(func() {
		cur, _, err := windows.Get(holder.BootstrapWindow)
		if err == nil && !cur.Closed.IsZero() {
			return
		}
		w := holder.Window{Name: holder.BootstrapWindow, Closed: time.Now().UTC()}
		if err := windows.Put(w); err != nil {
			logf("the bootstrap's one-off credential could not be asked away: %v", err)
			return
		}
		logf("the bootstrap's one-off credential is asked away: the holders serve publish")
	})
}

// learned are the proved callers whose build this process recorded as publishing through the seats.
var learned sync.Map

// learnPublishing records the fingerprint the proved caller's node runs of its build module as one that publishes
// through the seats (holder.PublishingKey), once per caller and process: the windows' driver in the program counts a
// node's have running it as publishing, closing the direct-write window when every node's build modules do.
func learnPublishing(windows kept[holder.Window], caller string) {
	node, module, ok := strings.Cut(caller, ".")
	if !ok || strings.HasPrefix(caller, "person.") {
		return
	}
	if _, done := learned.Load(caller); done {
		return
	}
	raw, err := stdio.Ask("mesh-delivery-view.node_haves", map[string]any{"node": node, "module": module})
	if err != nil {
		logf("the build of %s that published cannot be read (%v): the direct-write window counts it at its next publish", caller, err)
		return
	}
	var a struct {
		Keys map[string]struct {
			FP string `json:"fp"`
		} `json:"keys"`
	}
	if json.Unmarshal(raw, &a) != nil || a.Keys[caller].FP == "" {
		logf("node_haves names no build of %s: the direct-write window counts it at its next publish", caller)
		return
	}
	learnt, err := learnInto(stdio.State("windows"), module, a.Keys[caller].FP, learnAttempts)
	if err != nil {
		logf("the publishing build of %s is not kept: %v", caller, err)
		return
	}
	if learnt {
		logf("%s's build %s publishes through the seats", module, a.Keys[caller].FP)
	}
	learned.Store(caller, true)
}

// learnAttempts bounds learnInto's tries against other writers of the key.
const learnAttempts = 8

// casState is the part of a kept state learnInto writes through: a read with its revision, a create, an update at a
// revision (stdio.KeptState).
type casState interface {
	Get(key string) (*stdio.StateEntry, error)
	Create(key string, value any) (uint64, error)
	Update(key string, value any, revision uint64) (uint64, error)
}

// learnInto adds fp for module to the publishing builds kept at holder.PublishingKey with compare-and-set on the key's
// revision: two first publishes at once each read, add and write only if nobody wrote meanwhile, else read again, so
// neither is lost. It answers whether fp was new.
func learnInto(state casState, module, fp string, attempts int) (bool, error) {
	for i := 0; i < attempts; i++ {
		e, err := state.Get(holder.PublishingKey)
		if err != nil {
			return false, err
		}
		var cur holder.Publishing
		if e != nil {
			if err := json.Unmarshal(e.Value, &cur); err != nil {
				return false, fmt.Errorf("%s holds no publishing builds: %w", holder.PublishingKey, err)
			}
		}
		cur.Name = holder.PublishingKey
		if !cur.Learn(module, fp) {
			return false, nil
		}
		if e == nil {
			_, err = state.Create(holder.PublishingKey, cur)
		} else {
			_, err = state.Update(holder.PublishingKey, cur, e.Revision)
		}
		if err == nil {
			return true, nil
		}
		if !errors.Is(err, stdio.ErrChanged) {
			return false, err
		}
	}
	return false, fmt.Errorf("%s was written by another writer first %d times: %w", holder.PublishingKey, attempts, stdio.ErrChanged)
}

// mirrors is the state mirrors: an ask is created once; one already kept is the same copy, taken on already.
type mirrors struct{}

func (mirrors) Ask(key string, r holder.MirrorRequest) error {
	if _, err := stdio.State(holder.MirrorsState).Create(key, r); err != nil && !errors.Is(err, stdio.ErrChanged) {
		return err
	}
	return nil
}
func (mirrors) Claim(key string, r holder.MirrorRequest, revision uint64) (bool, error) {
	_, err := stdio.State(holder.MirrorsState).Update(key, r, revision)
	if errors.Is(err, stdio.ErrChanged) {
		return false, nil
	}
	return err == nil, err
}
func (mirrors) Delete(key string) error { return stdio.State(holder.MirrorsState).Delete(key) }

// followMirrors runs every copy `mirror` took on, on the state's watch: the asks kept when the bundle starts (a copy a
// restart cut) and each new one, one at a time. Each is claimed first (Verbs.Take), so a second follower copies none
// a first one took on.
func followMirrors(v *holder.Verbs) {
	id := make([]byte, 8)
	_, _ = rand.Read(id)
	self := "distribution-tools-" + hex.EncodeToString(id)
	<-stdio.Served()
	stdio.State(holder.MirrorsState).Follow("", func(ch stdio.StateChange) error {
		if ch.Op != stdio.OpPut {
			return nil
		}
		var r holder.MirrorRequest
		if err := json.Unmarshal(ch.Value, &r); err != nil {
			logf("the mirror ask %s is no ask (%v): dropped", ch.Key, err)
			return stdio.State(holder.MirrorsState).Delete(ch.Key)
		}
		return v.Take(ch.Key, r, ch.Revision, ch.Current, self)
	})
}

func arg[T any](args map[string]any) (T, error) {
	var v T
	clean := map[string]any{}
	for k, x := range args {
		if k != "_meta" {
			clean[k] = x
		}
	}
	b, _ := json.Marshal(clean)
	return v, json.Unmarshal(b, &v)
}

func str(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }

// tools are distribution's tools over v; each call that checks a caller reads the caller list on that call (callers:
// the config value as mesh-config resolves it now), so an operator's change of it holds from the next call on.
func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.Callers, error)) []stdio.Tool {
	now := func() (*holder.Verbs, error) {
		c, err := callers()
		if err != nil {
			return nil, fmt.Errorf("the caller list cannot be read, so no verb is answered: %w", err)
		}
		return v.WithCallers(c), nil
	}
	obj := func(props map[string]any, req ...string) map[string]any {
		return map[string]any{"type": "object", "required": req, "properties": props}
	}
	s := func(d string) map[string]any { return map[string]any{"type": "string", "description": d} }
	var out []stdio.Tool
	for _, seat := range holder.Seats {
		out = append(out,
			stdio.Tool{Name: seat + "." + holder.VerbPublish, Description: "Open a transfer of a publish (ADR 0611): its name, version, kind, digest and parts; answers the parts the holder lacks.",
				Input: obj(map[string]any{"transfer": s("the transfer's id"), "name": s("<module>/<artifact>"), "version": s("the builder's version"), "kind": s("image, bundle, archive or upstream"), "digest": s("sha256:…"), "parts": map[string]any{"type": "array"}}, "transfer", "name", "kind", "digest", "parts"),
				Run: func(args map[string]any) (any, error) {
					v, err := now()
					if err != nil {
						return nil, err
					}
					in, err := arg[reg.PublishInput](args)
					if err != nil {
						return nil, err
					}
					_, _ = v.DropIdle()
					a, err := v.Publish(seat, stdio.CallerOf(args), publishOf(in))
					if err == nil && closeBootstrap != nil {
						closeBootstrap()
					}
					if err == nil {
						learnPublishing(windows, stdio.CallerOf(args))
					}
					return publishAnswer(a), err
				}},
			stdio.Tool{Name: seat + "." + holder.VerbPublishPart, Description: "Send one chunk (at most 1 MiB) of one part of an open transfer, from the caller that opened it.",
				Input: obj(map[string]any{"transfer": s("its id"), "part": s("the part's digest"), "sequence": map[string]any{"type": "integer"}, "last": map[string]any{"type": "boolean"}, "data": s("base64")}, "transfer", "part", "sequence", "last", "data"),
				Run: func(args map[string]any) (any, error) {
					v, err := now()
					if err != nil {
						return nil, err
					}
					in, err := arg[reg.PublishPartInput](args)
					if err != nil {
						return nil, err
					}
					r, err := partOf(in)
					if err != nil {
						return nil, err
					}
					a, err := v.PublishPart(seat, stdio.CallerOf(args), r)
					return partAnswer(in, a), err
				}},
			stdio.Tool{Name: seat + "." + holder.VerbVersions, Description: "The versions the seat holds of a name.", Input: obj(map[string]any{"name": s("<module>/<artifact>")}, "name"),
				Run: func(args map[string]any) (any, error) {
					v, err := now()
					if err != nil {
						return nil, err
					}
					a, err := v.Versions(seat, stdio.CallerOf(args), str(args, "name"))
					if err != nil {
						return nil, err
					}
					return versionsAnswer(v, str(args, "kind"), a), nil
				}},
			stdio.Tool{Name: seat + "." + holder.VerbRemove, Description: "Remove a tagged version (the builder), or for retention an untagged one by its digest (the artifact sweep; an index with the platform manifests no tagged or held index names, and never a manifest such an index names); says removed.",
				Input: obj(map[string]any{"name": s("name"), "version": s("a tagged version"), "tag": s("its tag"), "digest": s("for retention"), "reason": s("why")}, "name", "reason"),
				Run: func(args map[string]any) (any, error) {
					v, err := now()
					if err != nil {
						return nil, err
					}
					in, err := arg[reg.RemoveInput](args)
					if err != nil {
						return nil, err
					}
					r := holder.RemoveRequest{Name: in.Name, Version: in.Version, Digest: in.Digest, Reason: in.Reason}
					if err := v.Remove(seat, stdio.CallerOf(args), r); err != nil {
						return nil, err
					}
					return reg.RemoveAnswer{Removed: true, Artifact: &reg.Artifact{Name: in.Name, Version: in.Version, Tag: in.Tag, Digest: in.Digest}}, nil
				}},
			stdio.Tool{Name: seat + "." + holder.VerbHold, Description: "Keep one untagged bundle, archive or copied base from retention (the artifact sweep alone); says held.",
				Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "reason": s("which of the sweep's rules keeps it")}, "name", "digest", "reason"),
				Run: func(args map[string]any) (any, error) {
					v, err := now()
					if err != nil {
						return nil, err
					}
					in, err := arg[reg.HoldInput](args)
					if err != nil {
						return nil, err
					}
					if err := v.HoldOne(seat, stdio.CallerOf(args), holder.HoldRequest{Name: in.Name, Digest: in.Digest, Why: in.Reason}); err != nil {
						return nil, err
					}
					return reg.HoldAnswer{Held: true, Since: time.Now().UTC().Format(time.RFC3339)}, nil
				}},
			stdio.Tool{Name: seat + "." + holder.VerbRelease, Description: "Let a hold go (the artifact sweep, or the operator's session through registry_holds); says released.",
				Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "reason": s("why it is let go")}, "name", "digest", "reason"),
				Run: func(args map[string]any) (any, error) {
					v, err := now()
					if err != nil {
						return nil, err
					}
					in, err := arg[reg.ReleaseInput](args)
					if err != nil {
						return nil, err
					}
					if err := v.Release(seat, stdio.CallerOf(args), holder.HoldRequest{Name: in.Name, Digest: in.Digest, Why: in.Reason}); err != nil {
						return nil, err
					}
					return reg.ReleaseAnswer{Released: true}, nil
				}},
		)
	}
	out = append(out,
		stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbTags, Description: "The tags of an image.", Input: obj(map[string]any{"name": s("name")}, "name"),
			Run: func(args map[string]any) (any, error) {
				v, err := now()
				if err != nil {
					return nil, err
				}
				a, err := v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name"))
				if err != nil {
					return nil, err
				}
				return tagsAnswer(v, a), nil
			}},
		stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Copy an outside image, pinned by its digest, into the store as upstream/<host>/<path> (every platform, only the parts it lacks): answers at once with the reference the copy will have, and says published (or refused, with why) on the seat's events when the copy ends; wait on that event (mesh-build-image alone).",
			Input: obj(map[string]any{"source": s("<host>/<path>"), "digest": s("sha256:…")}, "source", "digest"),
			Run: func(args map[string]any) (any, error) {
				v, err := now()
				if err != nil {
					return nil, err
				}
				in, err := arg[reg.MirrorInput](args)
				if err != nil {
					return nil, err
				}
				ref, err := v.Mirror(holder.ImageSeat, stdio.CallerOf(args), holder.MirrorRequest{Source: in.Source, Digest: in.Digest})
				if err != nil {
					return nil, err
				}
				return reg.MirrorAnswer{Digest: in.Digest, Reference: ref}, nil
			}},
		stdio.Tool{Name: holder.CallersTool, Description: "Who may call which verb of distribution's seats: the trusted config value callers as mesh-config resolves it on this machine now (design 51's list where unset). The program asks it at each wake of the windows, as it has no config of its own.",
			Input: obj(map[string]any{}),
			Run: func(map[string]any) (any, error) {
				c, err := callers()
				if err != nil {
					return nil, err
				}
				return map[string]any{"callers": c}, nil
			}},
		stdio.Tool{Name: "registry_health", Description: "distribution's health check, run by the runtime each minute: versions of each seat asked through the store's read side (the program's port 5000); fails naming the program when it does not answer.",
			Input: obj(map[string]any{}),
			Run: func(map[string]any) (any, error) {
				front, _ := url.Parse("http://127.0.0.1:5000")
				return holder.Health(&holder.OCI{Base: front}, finding)
			}},
		stdio.Tool{Name: "registry_incoming", Description: "Open transfers of incoming and its fill; with drop, drops one by hand (a hand act, why required).",
			Input: obj(map[string]any{"drop": s("a transfer id to drop"), "why": s("why it is dropped")}),
			Run: func(args map[string]any) (any, error) {
				v, err := now()
				if err != nil {
					return nil, err
				}
				idle, _ := v.DropIdle()
				if id := str(args, "drop"); id != "" {
					if str(args, "why") == "" {
						return nil, errors.New("a drop is a hand act: say why")
					}
					t, err := v.Drop(id)
					logf("transfer %s dropped by hand by %s: %s", id, stdio.CallerOf(args), str(args, "why"))
					return map[string]any{"dropped": t.ID}, err
				}
				all, err := v.Transfers.List()
				used, _ := v.Objects.Used()
				return map[string]any{"transfers": all, "used": used, "cap": holder.IncomingCap, "dropped_idle": idle}, err
			}},
		stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).",
			Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}),
			Run: func(args map[string]any) (any, error) {
				v, err := now()
				if err != nil {
					return nil, err
				}
				if name := str(args, "release"); name != "" {
					if str(args, "why") == "" {
						return nil, errors.New("a release by hand is a hand act: say why")
					}
					// The caller proved on this call (the operator's session) is carried on; distribution's own never.
					return map[string]any{"released": name}, v.Release(str(args, "seat"), stdio.CallerOf(args), holder.HoldRequest{Name: name, Digest: str(args, "digest"), Why: str(args, "why")})
				}
				return v.HeldNow()
			}},
	)
	for _, w := range []string{holder.OldNameWindow, holder.DirectWritesWindow} {
		name := "registry_old_name"
		if w == holder.DirectWritesWindow {
			name = "registry_direct_writes"
		}
		out = append(out, stdio.Tool{Name: name, Description: "Whether the " + w + " window is open and the nodes holding it, since when; with drop, drops a retired node from it by hand (a hand act, why required).",
			Input: obj(map[string]any{"drop": s("a retired node"), "why": s("why")}),
			Run: func(args map[string]any) (any, error) {
				cur, _, err := windows.Get(w)
				if err != nil {
					return nil, err
				}
				cur.Name = w
				if node := str(args, "drop"); node != "" {
					if cur, err = cur.Drop(node, stdio.CallerOf(args), str(args, "why"), time.Now().UTC()); err != nil {
						return nil, err
					}
					if err := windows.Put(cur); err != nil {
						return nil, err
					}
				}
				b, _, err := windows.Get(holder.BootstrapWindow)
				if err != nil {
					return nil, err
				}
				// The program removes the file when this window closes and writes it closed again with its removal's time.
				return map[string]any{"window": cur, "bootstrap_credential_discarded": !b.Closed.IsZero(), "bootstrap": b}, nil
			}})
	}
	return out
}

func main() {
	base, _ := url.Parse("http://127.0.0.1:5001")
	store := &holder.OCI{Base: base, Upstream: &holder.Upstream{}}
	h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding,
		Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState)}
	v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}}
	go followMirrors(v)
	windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }}
	closeBootstrap = bootstrapCloser(windows)
	if err := stdio.Serve("", tools(v, windows, holder.ConfiguredCallers)); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}