package main import ( "crypto/rand" "encoding/hex" "encoding/json" "fmt" "strings" "time" occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" "nats-tools/internal/keeper" ) // A person's acting tools (design 25 § 12.3; CAT-N8): nats_create and nats_apply (acts), nats_recreate and nats_drop // (dangerous, granted only by name). None holds a grant of the seat's acting verbs: each hands its request to the // keeper through the keeper's own state requests, stamped with the caller the tool runner proved, and reads its end // back. recreate and drop first raise the operator's ask, binding exactly that act; the keeper acts only on its warrant, // after a snapshot. // requestStore is the keeper's state requests, as this bundle reaches it. type requestStore struct { Put func(key string, r keeper.Request) error Get func(key string) (*keeper.Request, error) Ask func(a oc.Ask) error Now func() time.Time Wait func(d time.Duration) } // RequestsInState is the store through the runtime that launched this bundle. func RequestsInState() requestStore { s := stdio.State(keeper.StateRequests) return requestStore{ Put: func(key string, r keeper.Request) error { _, err := s.Put(key, r); return err }, Get: func(key string) (*keeper.Request, error) { e, err := s.Get(key) if err != nil || e == nil { return nil, err } var r keeper.Request return &r, json.Unmarshal(e.Value, &r) }, Ask: func(a oc.Ask) error { return stdio.SeatPublish(oc.AcceptSubjectAsk(keeper.Module), a, a.ID) }, Now: func() time.Time { return time.Now().UTC() }, Wait: time.Sleep, } } func requestID(now time.Time) string { var b [4]byte _, _ = rand.Read(b[:]) return now.Format("20060102t150405") + "-" + hex.EncodeToString(b[:]) } // askFor is the operator's ask for a recreate or a drop of one entry, its approve option binding exactly that act. func askFor(r keeper.Request, now time.Time) (oc.Ask, error) { binds, err := occ.ActDigest(r.Act()) if err != nil { return oc.Ask{}, err } what := map[string]string{keeper.VerbDrop: "remove", keeper.VerbRecreate: "rebuild"}[r.Verb] name := keeper.BusName(r.Module, r.Local) return oc.Ask{ID: "bus-" + r.ID, Who: occ.Operator, Expires: occ.FormatTime(now.Add(24 * time.Hour)), OnExpiry: "nothing is removed and the entry stays as it is", About: keeper.ConditionKey(map[string]string{keeper.VerbDrop: "orphaned", keeper.VerbRecreate: "drifted"}[r.Verb], r.Module, r.Local), Headline: fmt.Sprintf("%s's kept data %s: %s it?", r.Module, r.Local, what), Explanation: fmt.Sprintf("%s asks to %s %s on the bus, %s. A snapshot of it is taken first; its data is lost from the bus.", r.Caller, what, name, r.Why), Options: []oc.Option{ {ID: "go", Label: strings.ToUpper(what[:1]) + what[1:], Does: fmt.Sprintf("%s %s after a snapshot", what, name), Level: occ.Destroy, Binds: binds}, {ID: "keep", Label: "Keep it", Does: "nothing is removed", Level: occ.Acknowledge}, }}, nil } func requestTools(rs requestStore) []stdio.Tool { entry := map[string]any{"module": str("the module, e.g. mesh-knowledge"), "local": str("its local name, e.g. records"), "why": str("why, in plain words: kept with the act"), "request": str("a request this tool made before: its end is read, nothing new is asked")} tool := func(verb, description string) stdio.Tool { return stdio.Tool{Name: "nats_" + verb, Description: description, Input: entry, Run: func(a map[string]any) (any, error) { if id := arg(a, "request"); id != "" { r, err := rs.Get(id) if err != nil || r == nil { return nil, fmt.Errorf("no request %s is held (%v)", id, err) } return r, nil } r := keeper.Request{Verb: verb, Module: arg(a, "module"), Local: arg(a, "local"), Why: strings.TrimSpace(arg(a, "why")), // The caller the tool runner proved, never a name the arguments give. Caller: stdio.CallerOf(a), At: rs.Now()} if r.Module == "" || r.Local == "" || r.Why == "" { return nil, fmt.Errorf("nats_%s names one entry, a module and its local name, and says why", verb) } r.ID = requestID(r.At) if verb == keeper.VerbRecreate || verb == keeper.VerbDrop { ask, err := askFor(r, r.At) if err != nil { return nil, err } r.Ask = ask.ID if err := rs.Put(r.ID, r); err != nil { return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err) } if err := rs.Ask(ask); err != nil { return nil, fmt.Errorf("the operator's ask cannot be raised, so nothing will be removed: %w", err) } return map[string]any{"request": r.ID, "ask": ask.ID, "state": "the operator is asked; the keeper acts once they approve, after a snapshot", "read": fmt.Sprintf("nats_%s with request %s", verb, r.ID)}, nil } if err := rs.Put(r.ID, r); err != nil { return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err) } return awaitEnd(rs, r, "nats_"+verb), nil }} } firstView := stdio.Tool{Name: "nats_first_view_again", Description: "Ask the bus module's keeper to ask the indexer's indexer_bus_state, its first view of every accepted manifest's bus state, at once (design 25 § 12.2): by hand, beside its own backoff, recorded as a hand act; answers what the first view answered and its revision.", Input: map[string]any{"why": str("why, in plain words: kept with the act"), "request": str("a request made before: its end is read")}, Run: func(a map[string]any) (any, error) { if id := arg(a, "request"); id != "" { r, err := rs.Get(id) if err != nil || r == nil { return nil, fmt.Errorf("no request %s is held (%v)", id, err) } return r, nil } r := keeper.Request{Verb: keeper.VerbFirstViewAgain, Why: strings.TrimSpace(arg(a, "why")), Caller: stdio.CallerOf(a), At: rs.Now()} if r.Why == "" { return nil, fmt.Errorf("nats_first_view_again says why") } r.ID = requestID(r.At) if err := rs.Put(r.ID, r); err != nil { return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err) } return awaitEnd(rs, r, "nats_first_view_again"), nil }} return []stdio.Tool{ firstView, tool(keeper.VerbCreate, "Create one entry of a module's kept data on the bus now, as its accepted manifest asks for it (design 25 § 12.3): handed to the bus module's keeper, which alone may create, and recorded as a hand act."), tool(keeper.VerbApply, "Apply what a module's manifest asks of one entry where the bus holds less (a larger cap, a longer age): only a difference that grows; handed to the keeper and recorded as a hand act."), tool(keeper.VerbRecreate, "Rebuild one entry whose settings drifted in a way that does not grow: the operator is asked first, and the keeper acts only on their warrant, after a snapshot of the entry. Its data is lost from the bus (the snapshot keeps it). Dangerous."), tool(keeper.VerbDrop, "Remove one orphan, an entry no accepted manifest asks for: the operator is asked first, and the keeper acts only on their warrant, after a snapshot of the entry. Dangerous."), } } // awaitEnd reads a request's end back: the keeper acts on its watch of requests at once, so the end is read for at most // half a minute, and otherwise how to read it later is answered. func awaitEnd(rs requestStore, r keeper.Request, tool string) any { for wait := 100 * time.Millisecond; wait < 20*time.Second; wait *= 2 { rs.Wait(wait) if got, err := rs.Get(r.ID); err == nil && got != nil && !got.Done.IsZero() { return got } } return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", "read": fmt.Sprintf("%s with request %s", tool, r.ID)} }