package buscompose import ( "fmt" "testing" "time" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/seatdef" ) // Defends issue 1790 (ADR 0547 ยง5): the grants part creates a declared bucket and stream as the controller asserts them. func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"}) zero, seven, sixtyFour := 0, 7, 64 defs := map[string]seatdef.Definition{ "node-engine": {Seat: "node-engine", Events: []seatdef.Message{ {Name: "running", Subject: "running..", Keep: &seatdef.Keep{Object: true, MaxAgeDays: &zero, LastPerSubject: true}}, {Name: "composed"}}}, "mesh-delivery": {Seat: "mesh-delivery", Events: []seatdef.Message{ {Name: "owed", Subject: "owed..", Keep: &seatdef.Keep{Object: true, MaxAgeDays: &seven, MaxMiB: &sixtyFour}}}}, "mesh-catalog": {Seat: "mesh-catalog", Events: []seatdef.Message{{Name: "registration-changed", Keep: &seatdef.Keep{Object: true}}}}, } kept := KeptEventsOf(defs) byStream := map[string]jetstream.StreamConfig{} for _, k := range kept { byStream[k.Stream] = k.StreamConfig() } running, owed, named := byStream["SEAT_node-engine_running"], byStream["SEAT_mesh-delivery_owed"], byStream["SEAT_mesh-catalog_registration-changed"] rest, heard := SplitKept(kept, []string{"mesh-catalog.manifest-accepted", "node-engine.running.*.*", "node-engine.running"}) heardConsumerCfg := ConsumerConfig(heardConsumer(Principal{Kind: KindModule, Node: "novox", Module: "mesh-grants"}, heard[0])) keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true}) queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32, WorkQueue: true, Duplicates: 1}) for _, row := range []struct { what string got, want any }{ {"bucket name", plain.Name, "KV_mesh-vault_bus-accounts"}, {"bucket subjects", plain.Subjects[0], "$KV.mesh-vault_bus-accounts.>"}, {"bucket history defaults to one", plain.MaxMsgsPerSubject, int64(1)}, {"bucket cap", plain.MaxBytes, int64(StateMaxBytes)}, {"bucket value cap", plain.MaxMsgSize, int32(StateMaxValueBytes)}, {"bucket duplicate window", plain.Duplicates, 2 * time.Minute}, {"bucket discards new", plain.Discard, jetstream.DiscardNew}, {"bucket direct and rollup, no delete", plain.AllowDirect && plain.AllowRollup && plain.DenyDelete, true}, {"bucket without key ttl", plain.AllowMsgTTL, false}, {"history", keyed.MaxMsgsPerSubject, int64(5)}, {"age", keyed.MaxAge, time.Minute}, {"duplicate window under the age", keyed.Duplicates, time.Minute}, {"key ttl markers", keyed.SubjectDeleteMarkerTTL, KeyTTLMarkerTTL}, {"key ttl", keyed.AllowMsgTTL, true}, {"queue retention", queue.Retention, jetstream.WorkQueuePolicy}, {"queue subjects", queue.Subjects[0], "mesh.stream.b.event.ask"}, {"queue age", queue.MaxAge, 7 * 24 * time.Hour}, {"queue cap", queue.MaxBytes, int64(32 << 20)}, {"queue discards new", queue.Discard, jetstream.DiscardNew}, {"queue duplicates", queue.Duplicates, time.Hour}, {"queue metadata", queue.Metadata[WorkQueueMetadata] + queue.Metadata[RestrictedMetadata], "truetrue"}, {"kept events are three", len(kept), 3}, {"kept stream name", running.Name, "SEAT_node-engine_running"}, {"a kept stream takes no subject of the shared stream", len(running.Subjects), 0}, {"a kept stream sources the shared stream by the event's subject", running.Sources[0].Name + " " + running.Sources[0].FilterSubject, "EVENTS mesh.seat.node-engine.event.running.*.*"}, {"last per subject holds one message a subject", running.MaxMsgsPerSubject, int64(1)}, {"last per subject has no age", running.MaxAge, time.Duration(0)}, {"a keep naming nothing is 90 days within 16 MiB, dropping its oldest", fmt.Sprint(named.MaxAge, named.MaxBytes, named.Discard == jetstream.DiscardOld), fmt.Sprint(90*24*time.Hour, int64(16<<20), true)}, {"the stream kept for a week", owed.MaxAge, 7 * 24 * time.Hour}, {"the stream kept within its size", owed.MaxBytes, int64(64 << 20)}, {"its name is the bus's", owed.Name, "SEAT_mesh-delivery_owed"}, {"a kept consume leaves the shared stream's", len(rest), 1}, {"both namings of one event are one durable on its stream", heard[0].Stream + " " + fmt.Sprint(heard[0].Filters), "SEAT_node-engine_running [mesh.seat.node-engine.event.running.*.*]"}, {"its durable starts at the last message of each subject", heardConsumerCfg.DeliverPolicy, jetstream.DeliverLastPerSubjectPolicy}, {"its durable is the module's on the node", heardConsumerCfg.Durable, "novox_mesh-grants"}, } { if row.got != row.want { t.Errorf("%s: %v, want %v", row.what, row.got, row.want) } } }