diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go index ba1c0610..db7552df 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go @@ -10,8 +10,8 @@ import ( "testing" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-view/internal/view" diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index d2dc23a5..24999173 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -12,9 +12,9 @@ import ( "mesh-delivery-view/internal/view" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/provisions" "git.novox.be/novox/mesh-sdk/go/views" @@ -426,7 +426,7 @@ func main() { } // view_first_views and view_first_view_again: where the view's process stands with its first views of the indexer's and // the assigner's states (ADR 0615 ยง6), and a re-read by hand. - tools := append(Tools(r, u), views.StatusTools(delivery.ViewModule, "view_first_views", "view_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) + tools := append(Tools(r, u), views.StatusTools(delivery.ViewModule, "view_first_views", "view_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index a3f84601..69f5a9dd 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -16,8 +16,8 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" @@ -32,7 +32,7 @@ import ( func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { t.Helper() for _, o := range ownersFollowed { - viewstest.Serve(t, ctx, js, nc, o.owner, o.state, o.tool, o.stream, "n_"+delivery.ViewModule) + viewstest.Serve(t, ctx, js, nc, o.owner, o.state, o.tool, "n_"+delivery.ViewModule) } } @@ -750,7 +750,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() - events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}}) + events, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>"}}) if err != nil { t.Fatal(err) } @@ -801,7 +801,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { t.Fatal(err) } time.Sleep(time.Second) - if info, err := events.Info(ctx); err != nil || info.State.Msgs != 0 { + if info, err := events.Info(ctx, jetstream.WithSubjectFilter("mesh.mod.mesh-delivery-view.event.>")); err != nil || len(info.State.Subjects) != 0 { t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State) } if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", good); err != nil { @@ -847,7 +847,7 @@ func TestProcessWholeSaysModulesInMotion(t *testing.T) { } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() - events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}}) + events, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>"}}) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 05b51377..c8ccf17e 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + sdkfindings "git.novox.be/novox/mesh-delivery/sdk/findings" "io" "log" "net/http" @@ -117,9 +118,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer calls.Close() reader := &views.ProcessReader{Module: delivery.ViewModule, Call: calls.Call, Logf: logger.Printf} - reader.Findings = process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf).KV + findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) + reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } for _, o := range ownersFollowed { - if _, err := reader.Follow(o.owner, o.state, o.tool, o.stream); err != nil { + if _, err := reader.Follow(o.owner, o.state, o.tool); err != nil { return err } } @@ -180,15 +182,15 @@ var ( ) // ownersFollowed are the indexer's and the assigner's states the view's process follows by their first views, each by its -// owner's read tool (as the view's invokes names it) and views family (as the owner's manifest's `streams` names it); the +// owner's read tool (as the view's invokes names it) and views family (its events on the shared events stream); the // view's tools ask the other six through the owners' read tools on each call. -var ownersFollowed = []struct{ owner, state, tool, stream string }{ - {delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule + ".assigner_read_placement", "assigner-views"}, - {delivery.AssignerModule, "wants", delivery.AssignerModule + ".assigner_read_wants", "assigner-views"}, - {delivery.AssignerModule, "asks", delivery.AssignerModule + ".assigner_read_asks", "assigner-views"}, - {delivery.IndexerModule, "candidates", delivery.IndexerModule + ".indexer_read_candidates", "indexer-views"}, - {delivery.IndexerModule, "modules", delivery.IndexerModule + ".indexer_read_modules", "indexer-views"}, - {delivery.IndexerModule, "taken", delivery.IndexerModule + ".indexer_read_taken", "indexer-views"}, +var ownersFollowed = []struct{ owner, state, tool string }{ + {delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule + ".assigner_read_placement"}, + {delivery.AssignerModule, "wants", delivery.AssignerModule + ".assigner_read_wants"}, + {delivery.AssignerModule, "asks", delivery.AssignerModule + ".assigner_read_asks"}, + {delivery.IndexerModule, "candidates", delivery.IndexerModule + ".indexer_read_candidates"}, + {delivery.IndexerModule, "modules", delivery.IndexerModule + ".indexer_read_modules"}, + {delivery.IndexerModule, "taken", delivery.IndexerModule + ".indexer_read_taken"}, } // buckets opens what the view watches: an owner's state as its picture (firstview), any other bucket on the bus. diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go index 120a2944..f045cac0 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go @@ -7,7 +7,7 @@ import ( "strings" "testing" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" ) func TestRunNeedsItsCredential(t *testing.T) { diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 2ce3d50d..fce17415 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,6 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 @@ -23,8 +24,8 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 - git.novox.be/novox/mesh-contracts/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-20261104174344-ae5266e77639 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-20261104174344-ae5266e77639 ) require ( diff --git a/modules/mesh-delivery-view/internal/view/answers_test.go b/modules/mesh-delivery-view/internal/view/answers_test.go index b936c375..4717494b 100644 --- a/modules/mesh-delivery-view/internal/view/answers_test.go +++ b/modules/mesh-delivery-view/internal/view/answers_test.go @@ -10,8 +10,8 @@ import ( "testing" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-view/internal/view/asks_test.go b/modules/mesh-delivery-view/internal/view/asks_test.go index 6fbf270b..cdc58a24 100644 --- a/modules/mesh-delivery-view/internal/view/asks_test.go +++ b/modules/mesh-delivery-view/internal/view/asks_test.go @@ -9,8 +9,8 @@ import ( "testing" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-view/internal/view/holds.go b/modules/mesh-delivery-view/internal/view/holds.go index a2b2f524..cd3aad09 100644 --- a/modules/mesh-delivery-view/internal/view/holds.go +++ b/modules/mesh-delivery-view/internal/view/holds.go @@ -10,7 +10,7 @@ import ( "strings" "time" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 0efa7e66..bfc21be3 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -14,8 +14,8 @@ import ( "sync/atomic" "time" - "git.novox.be/novox/mesh-delivery/sdk/buildlog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" + "git.novox.be/novox/mesh-delivery/sdk/buildlog" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" ) diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 44a464a6..444847ad 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -11,8 +11,8 @@ import ( "sync" "time" - occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" - oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/delivery" )