web: reorder status buttons

Makefile

@@ -0,0 +1,8 @@
+.PHONY: maps-service web validation
+maps-service:
+	DOCKER_BUILDKIT=1 docker build . -f Containerfile -t maps-service \
+		--secret id=netrc,src=/home/quat/.netrc
+web:
+	docker build web -f web/Containerfile -t maps-service-web
+validation:
+	docker build validation -f validation/Containerfile -t maps-service-validation

compose.yaml

@@ -3,32 +3,23 @@ networks:
   maps-service-network:
     driver: bridge
 
-secrets:
-  netrc:
-    file: /home/quat/.netrc
-
 services:
   nats:
     image: docker.io/nats:latest
     container_name: nats
-    command: ["-js"]
+    command: ["-js"] #"-DVV"
     networks:
       - maps-service-network
-    ports:
-      - "4222:4222"
 
-  maps-service:
+  mapsservice:
-    build:
+    image:
-      secrets:
+      maps-service
-        - netrc
+    container_name: mapsservice
-      context: .
-      dockerfile: Containerfile
-    container_name: maps-service
     command: [
       # debug
       "--debug","serve",
       # http service port
-      "--port","8081",
+      "--port","8082",
       # postgres
       "--pg-host","10.0.0.29",
       "--pg-port","5432",
@@ -37,38 +28,92 @@ services:
       "--pg-password","happypostgresuser",
       # other hosts
       "--nats-host","nats:4222",
-      "--auth-rpc-host","localhost:8090"
+      "--auth-rpc-host","authrpc:8081"
     ]
     depends_on:
+      - authrpc
       - nats
     networks:
       - maps-service-network
     ports:
-      - "8081:8081"
+      - "8082:8082"
 
   web:
-    build:
+    image:
-      context: web
+      maps-service-web
-      dockerfile: Containerfile
     networks:
       - maps-service-network
     ports:
       - "3000:3000"
 
   validation:
-    build:
+    image:
-      context: validation
+      maps-service-validation
-      dockerfile: Containerfile
     container_name: validation
     environment:
-      - RBXCOOKIE=RBXCOOKIE
+      - RBXCOOKIE
-      - API_HOST=http://localhost:8081
+      - API_HOST=http://mapsservice:8082
       - NATS_HOST=nats:4222
-      - DATA_HOST=http://localhost:9000
+      - DATA_HOST=http://dataservice:9000
     depends_on:
       - nats
-      # note: this races the maps-service which creates a nats stream
+      # note: this races the mapsservice which creates a nats stream
       # the validation will panic if the nats stream is not created
-      - maps-service
+      - mapsservice
+      - dataservice
     networks:
       - maps-service-network
+
+  dataservice:
+    image: registry.itzana.me/strafesnet/data-service:master
+    container_name: dataservice
+    environment:
+      - DEBUG=true
+      - PG_HOST=10.0.0.29
+      - PG_PORT=5432
+      - PG_USER=quat
+      - PG_DB=data
+      - PG_PASS=happypostgresuser
+    networks:
+      - maps-service-network
+
+  authredis:
+    image: docker.io/redis:latest
+    container_name: authredis
+    volumes:
+      - redis-data:/data
+    command: ["redis-server", "--appendonly", "yes"]
+    networks:
+      - maps-service-network
+
+  authrpc:
+    image: registry.itzana.me/strafesnet/auth-service:master
+    container_name: authrpc
+    command: ["serve", "rpc"]
+    environment:
+      - REDIS_ADDR=authredis:6379
+    env_file:
+      - ../auth-compose/auth-service.env
+    depends_on:
+      - authredis
+    networks:
+      - maps-service-network
+    logging:
+      driver: "none"
+
+  auth-web:
+    image: registry.itzana.me/strafesnet/auth-service:master
+    command: ["serve", "web"]
+    environment:
+      - REDIS_ADDR=authredis:6379
+    env_file:
+      - ../auth-compose/auth-service.env
+    depends_on:
+      - authredis
+    networks:
+      - maps-service-network
+    ports:
+      - "8080:8080"
+
+volumes:
+  redis-data:

pkg/service/submissions.go

@@ -313,7 +313,7 @@ func (svc *Service) ActionSubmissionTriggerPublish(ctx context.Context, params a
 			return err
 		}
 
-		svc.Nats.Publish("maptest.submissions.publish.new", []byte(j))
+		svc.Nats.Publish("maptest.submissions.publishnew", []byte(j))
 	} else {
 		// this is a map fix
 		publish_fix_request := model.PublishFixRequest{
@@ -328,7 +328,7 @@ func (svc *Service) ActionSubmissionTriggerPublish(ctx context.Context, params a
 			return err
 		}
 
-		svc.Nats.Publish("maptest.submissions.publish.fix", []byte(j))
+		svc.Nats.Publish("maptest.submissions.publishfix", []byte(j))
 	}
 
 	return nil

validation/Cargo.lock

@@ -171,8 +171,8 @@ dependencies = [
  "pin-project-lite",
  "rustversion",
  "serde",
- "sync_wrapper 1.0.2",
+ "sync_wrapper",
- "tower 0.5.1",
+ "tower 0.5.2",
  "tower-layer",
  "tower-service",
 ]
@@ -192,7 +192,7 @@ dependencies = [
  "mime",
  "pin-project-lite",
  "rustversion",
- "sync_wrapper 1.0.2",
+ "sync_wrapper",
  "tower-layer",
  "tower-service",
 ]
@@ -287,9 +287,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.2.2"
+version = "1.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f34d93e62b03caf570cccc334cbc6c2fceca82f39211051345108adcba3eebdc"
+checksum = "9157bbaa6b165880c27a4293a474c91cdcf265cc68cc829bf10be0964a391caf"
 dependencies = [
  "shlex",
 ]
@@ -302,9 +302,9 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
 [[package]]
 name = "chrono"
-version = "0.4.38"
+version = "0.4.39"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a21f936df1771bf62b77f047b726c4625ff2e8aa607c01ec06e5a05bd8463401"
+checksum = "7e36cc9d416881d2e24f9a963be5fb1cd90966419ac844274161d10488b3e825"
 dependencies = [
  "android-tzdata",
  "iana-time-zone",
@@ -510,9 +510,9 @@ dependencies = [
 
 [[package]]
 name = "fastrand"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "486f806e73c5707928240ddc295403b1b93c96a02038563881c4a2fd84b81ac4"
+checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
 
 [[package]]
 name = "fiat-crypto"
@@ -709,9 +709,9 @@ checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
 
 [[package]]
 name = "http"
-version = "1.1.0"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21b9ddb458710bc376481b842f5da65cdf31522de232c1ca8146abce2a358258"
+checksum = "f16ca2af56261c99fba8bac40a10251ce8188205a4c448fbb745a2e4daa76fea"
 dependencies = [
  "bytes",
  "fnv",
@@ -1044,9 +1044,9 @@ checksum = "d75a2a4b1b190afb6f5425f10f6a8f959d2ea0b9c2b1d79553551850539e4674"
 
 [[package]]
 name = "js-sys"
-version = "0.3.74"
+version = "0.3.76"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a865e038f7f6ed956f788f0d7d60c541fff74c7bd74272c5d4cf15c63743e705"
+checksum = "6717b6b5b077764fb5966237269cb3c64edddde4b14ce42647430a78ced9e7b7"
 dependencies = [
  "once_cell",
  "wasm-bindgen",
@@ -1060,9 +1060,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
 
 [[package]]
 name = "libc"
-version = "0.2.167"
+version = "0.2.168"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09d6582e104315a817dff97f75133544b2e094ee22447d2acf4a74e189ba06fc"
+checksum = "5aaeb2981e0606ca11d79718f8bb01164f1d6ed75080182d3abf017e6d244b6d"
 
 [[package]]
 name = "linux-raw-sys"
@@ -1477,9 +1477,9 @@ dependencies = [
 
 [[package]]
 name = "rbx_asset"
-version = "0.2.3"
+version = "0.2.5"
 source = "sparse+https://git.itzana.me/api/packages/strafesnet/cargo/"
-checksum = "15542e7184e4b18e578308308cee72abcb2bfcb1d6c162f1eea50ec6cff7e3ac"
+checksum = "dcf243f46bd41b3880a27278177a3f9996f95ab231d9a04345ad9dd381c3a54a"
 dependencies = [
  "chrono",
  "flate2",
@@ -1627,7 +1627,7 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_urlencoded",
- "sync_wrapper 1.0.2",
+ "sync_wrapper",
  "system-configuration",
  "tokio",
  "tokio-native-tls",
@@ -1705,22 +1705,22 @@ dependencies = [
 
 [[package]]
 name = "rustix"
-version = "0.38.41"
+version = "0.38.42"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7f649912bc1495e167a6edee79151c84b1bad49748cb4f1f1167f459f6224f6"
+checksum = "f93dc38ecbab2eb790ff964bb77fa94faf256fd3e73285fd7ba0903b76bedb85"
 dependencies = [
  "bitflags 2.6.0",
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
 name = "rustls"
-version = "0.23.19"
+version = "0.23.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "934b404430bb06b3fae2cba809eb45a1ab1aecd64491213d7c3301b88393f8d1"
+checksum = "5065c3f250cbd332cd894be57c40fa52387247659b14a2d6041d121547903b1b"
 dependencies = [
  "once_cell",
  "ring",
@@ -1766,9 +1766,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-pki-types"
-version = "1.10.0"
+version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "16f1201b3c9a7ee8039bcadc17b7e605e2945b27eee7631788c1bd2b0643674b"
+checksum = "d2bf47e6ff922db3825eb750c4e2ff784c6ff8fb9e13046ef6a1d1c5401b0b37"
 
 [[package]]
 name = "rustls-webpki"
@@ -1840,24 +1840,24 @@ dependencies = [
 
 [[package]]
 name = "semver"
-version = "1.0.23"
+version = "1.0.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61697e0a1c7e512e84a621326239844a24d8207b4669b41bc18b32ea5cbf988b"
+checksum = "3cb6eb87a131f756572d7fb904f6e7b68633f09cca868c5df1c4b8d1a694bbba"
 
 [[package]]
 name = "serde"
-version = "1.0.215"
+version = "1.0.216"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6513c1ad0b11a9376da888e3e0baa0077f1aed55c17f50e7b2397136129fb88f"
+checksum = "0b9781016e935a97e8beecf0c933758c97a5520d32930e460142b4cd80c6338e"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.215"
+version = "1.0.216"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ad1e866f866923f252f05c889987993144fb74e722403468a4ebd70c3cd756c0"
+checksum = "46f859dbbf73865c6627ed570e78961cd3ac92407a2d117204c49232485da55e"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1925,6 +1925,15 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
 
+[[package]]
+name = "signal-hook-registry"
+version = "1.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9e9e0b4211b72e7b8b6e85c807d36c212bdb33ea8587f7569562a84df5465b1"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "signatory"
 version = "0.27.1"
@@ -2017,12 +2026,6 @@ dependencies = [
  "unicode-ident",
 ]
 
-[[package]]
-name = "sync_wrapper"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160"
-
 [[package]]
 name = "sync_wrapper"
 version = "1.0.2"
@@ -2099,9 +2102,9 @@ dependencies = [
 
 [[package]]
 name = "time"
-version = "0.3.36"
+version = "0.3.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5dfd88e563464686c916c7e46e623e520ddc6d79fa6641390f2e3fa86e83e885"
+checksum = "35e7868883861bd0e56d9ac6efcaaca0d6d5d82a2a7ec8209ff492c07cf37b21"
 dependencies = [
  "deranged",
  "itoa",
@@ -2120,9 +2123,9 @@ checksum = "ef927ca75afb808a4d64dd374f00a2adf8d0fcff8e7b184af886c3c87ec4a3f3"
 
 [[package]]
 name = "time-macros"
-version = "0.2.18"
+version = "0.2.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f252a68540fde3a3877aeea552b832b40ab9a69e318efd078774a01ddee1ccf"
+checksum = "2834e6017e3e5e4b9834939793b282bc03b37a3336245fa820e35e233e2a85de"
 dependencies = [
  "num-conv",
  "time-core",
@@ -2140,15 +2143,16 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.41.1"
+version = "1.42.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "22cfb5bee7a6a52939ca9224d6ac897bb669134078daa8735560897f69de4d33"
+checksum = "5cec9b21b0450273377fc97bd4c33a8acffc8c996c987a7c5b319a0083707551"
 dependencies = [
  "backtrace",
  "bytes",
  "libc",
  "mio",
  "pin-project-lite",
+ "signal-hook-registry",
  "socket2",
  "tokio-macros",
  "windows-sys 0.52.0",
@@ -2177,12 +2181,11 @@ dependencies = [
 
 [[package]]
 name = "tokio-rustls"
-version = "0.26.0"
+version = "0.26.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c7bc40d0e5a97695bb96e27995cd3a08538541b0a846f65bba7a359f36700d4"
+checksum = "5f6d0975eaace0cf0fcadee4e4aaa5da15b5c079146f2cffb67c113be122bf37"
 dependencies = [
  "rustls",
- "rustls-pki-types",
  "tokio",
 ]
 
@@ -2199,9 +2202,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-util"
-version = "0.7.12"
+version = "0.7.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61e7c3654c13bcd040d4a03abee2c75b1d14a37b423cf5a813ceae1cc903ec6a"
+checksum = "d7fcaa8d55a2bdd6b83ace262b016eca0d79ee02818c5c1bcdf0305114081078"
 dependencies = [
  "bytes",
  "futures-core",
@@ -2283,14 +2286,14 @@ dependencies = [
 
 [[package]]
 name = "tower"
-version = "0.5.1"
+version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2873938d487c3cfb9aed7546dc9f2711d867c9f90c46b889989a2cb84eba6b4f"
+checksum = "d039ad9159c98b70ecfd540b2573b97f7f52c3e8d9f8ad57a24b916a536975f9"
 dependencies = [
  "futures-core",
  "futures-util",
  "pin-project-lite",
- "sync_wrapper 0.1.2",
+ "sync_wrapper",
  "tower-layer",
  "tower-service",
 ]
@@ -2431,9 +2434,9 @@ checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.97"
+version = "0.2.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d15e63b4482863c109d70a7b8706c1e364eb6ea449b201a76c5b89cedcec2d5c"
+checksum = "a474f6281d1d70c17ae7aa6a613c87fce69a127e2624002df63dcb39d6cf6396"
 dependencies = [
  "cfg-if",
  "once_cell",
@@ -2442,13 +2445,12 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-backend"
-version = "0.2.97"
+version = "0.2.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8d36ef12e3aaca16ddd3f67922bc63e48e953f126de60bd33ccc0101ef9998cd"
+checksum = "5f89bb38646b4f81674e8f5c3fb81b562be1fd936d84320f3264486418519c79"
 dependencies = [
  "bumpalo",
  "log",
- "once_cell",
  "proc-macro2",
  "quote",
  "syn",
@@ -2457,9 +2459,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.47"
+version = "0.4.49"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9dfaf8f50e5f293737ee323940c7d8b08a66a95a419223d9f41610ca08b0833d"
+checksum = "38176d9b44ea84e9184eff0bc34cc167ed044f816accfe5922e54d84cf48eca2"
 dependencies = [
  "cfg-if",
  "js-sys",
@@ -2470,9 +2472,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.97"
+version = "0.2.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "705440e08b42d3e4b36de7d66c944be628d579796b8090bfa3471478a2260051"
+checksum = "2cc6181fd9a7492eef6fef1f33961e3695e4579b9872a6f7c83aee556666d4fe"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -2480,9 +2482,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.97"
+version = "0.2.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "98c9ae5a76e46f4deecd0f0255cc223cfa18dc9b261213b8aa0c7b36f61b3f1d"
+checksum = "30d7a95b763d3c45903ed6c81f156801839e5ee968bb07e534c44df0fcd330c2"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2493,15 +2495,15 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.97"
+version = "0.2.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6ee99da9c5ba11bd675621338ef6fa52296b76b83305e9b6e5c77d4c286d6d49"
+checksum = "943aab3fdaaa029a6e0271b35ea10b72b943135afe9bffca82384098ad0e06a6"
 
 [[package]]
 name = "web-sys"
-version = "0.3.74"
+version = "0.3.76"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a98bc3c33f0fe7e59ad7cd041b89034fa82a7c2d4365ca538dda6cdaf513863c"
+checksum = "04dd7223427d52553d3702c004d3b2fe07c148165faa56313cb00211e31c12bc"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -2642,9 +2644,9 @@ checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51"
 
 [[package]]
 name = "xml-rs"
-version = "0.8.23"
+version = "0.8.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "af310deaae937e48a26602b730250b4949e125f468f11e6990be3e5304ddd96f"
+checksum = "ea8b391c9a790b496184c29f7f93b9ed5b16abb306c05415b68bcc16e4d06432"
 
 [[package]]
 name = "yoke"

validation/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2021"
 api = { path = "api" }
 async-nats = "0.38.0"
 futures = "0.3.31"
-rbx_asset = { version = "0.2.3", registry = "strafesnet" }
+rbx_asset = { version = "0.2.5", registry = "strafesnet" }
 rbx_binary = { version = "0.7.4", registry = "strafesnet"}
 rbx_dom_weak = { version = "2.9.0", registry = "strafesnet"}
 rbx_reflection_database = { version = "0.2.12", registry = "strafesnet"}
@@ -16,5 +16,5 @@ rust-grpc = { version = "1.0.3", registry = "strafesnet" }
 serde = { version = "1.0.215", features = ["derive"] }
 serde_json = "1.0.133"
 siphasher = "1.0.1"
-tokio = { version = "1.41.1", features = ["macros", "rt-multi-thread", "fs"] }
+tokio = { version = "1.41.1", features = ["macros", "rt-multi-thread", "fs", "signal"] }
 tonic = "0.12.3"

validation/src/main.rs

@@ -1,8 +1,10 @@
-mod types;
+use futures::StreamExt;
+
 mod nats_types;
 mod validator;
 mod publish_new;
 mod publish_fix;
+mod message_handler;
 
 #[allow(dead_code)]
 #[derive(Debug)]
@@ -10,7 +12,8 @@ pub enum StartupError{
 	API(api::ReqwestError),
 	NatsConnect(async_nats::ConnectError),
 	NatsGetStream(async_nats::jetstream::context::GetStreamError),
-	NatsStartup(types::NatsStartupError),
+	NatsConsumer(async_nats::jetstream::stream::ConsumerError),
+	NatsStream(async_nats::jetstream::consumer::StreamError),
 	GRPCConnect(tonic::transport::Error),
 }
 impl std::fmt::Display for StartupError{
@@ -20,8 +23,13 @@ impl std::fmt::Display for StartupError{
 }
 impl std::error::Error for StartupError{}
 
+// annoying mile-long type
+pub type MapsServiceClient=rust_grpc::maps::maps_service_client::MapsServiceClient<tonic::transport::channel::Channel>;
+
 pub const GROUP_STRAFESNET:u64=6980477;
 
+pub const PARALLEL_REQUESTS:usize=16;
+
 #[tokio::main]
 async fn main()->Result<(),StartupError>{
 	// talk to roblox through STRAFESNET_CI2 account
@@ -34,29 +42,57 @@ async fn main()->Result<(),StartupError>{
 
 	// nats
 	let nats_host=std::env::var("NATS_HOST").expect("NATS_HOST env required");
-	let nasty=async_nats::connect(nats_host).await.map_err(StartupError::NatsConnect)?;	// use nats jetstream
+	let nats_fut=async{
-	let stream=async_nats::jetstream::new(nasty)
+		let nasty=async_nats::connect(nats_host).await.map_err(StartupError::NatsConnect)?;
-		.get_stream("maptest").await.map_err(StartupError::NatsGetStream)?;
+		// use nats jetstream
+		async_nats::jetstream::new(nasty)
+			.get_stream("maptest").await.map_err(StartupError::NatsGetStream)?
+			.get_or_create_consumer("validation",async_nats::jetstream::consumer::pull::Config{
+				name:Some("validation".to_owned()),
+				durable_name:Some("validation".to_owned()),
+				filter_subject:"maptest.submissions.>".to_owned(),
+				..Default::default()
+			}).await.map_err(StartupError::NatsConsumer)?
+			.messages().await.map_err(StartupError::NatsStream)
+	};
 
 	// data-service grpc for creating map entries
 	let data_host=std::env::var("DATA_HOST").expect("DATA_HOST env required");
-	let maps_grpc=crate::types::MapsServiceClient::connect(data_host).await.map_err(StartupError::GRPCConnect)?;
+	let message_handler_fut=async{
+		let maps_grpc=crate::MapsServiceClient::connect(data_host).await.map_err(StartupError::GRPCConnect)?;
+		Ok(message_handler::MessageHandler::new(cookie_context,api,maps_grpc))
+	};
 
-	// connect to nats
+	// Create a signal listener for SIGTERM
-	let (publish_new,publish_fix,validator)=tokio::try_join!(
+	let mut sig_term=tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("Failed to create SIGTERM signal listener");
-		publish_new::Publisher::new(stream.clone(),cookie_context.clone(),api.clone(),maps_grpc),
-		publish_fix::Publisher::new(stream.clone(),cookie_context.clone(),api.clone()),
-		// clone nats here because it's dropped within the function scope,
-		// meanining the last reference is dropped...
-		validator::Validator::new(stream.clone(),cookie_context,api)
-	).map_err(StartupError::NatsStartup)?;
 
-	// publisher threads
+	// run futures
-	tokio::spawn(publish_new.run());
+	let (mut messages,message_handler)=tokio::try_join!(nats_fut,message_handler_fut)?;
-	tokio::spawn(publish_fix.run());
 
-	// run validator on the main thread indefinitely
+	// process up to PARALLEL_REQUESTS in parallel
-	validator.run().await;
+	let main_loop=async move{
+		static SEM:tokio::sync::Semaphore=tokio::sync::Semaphore::const_new(PARALLEL_REQUESTS);
+		// use memory leak to make static lifetime
+		let message_handler=Box::leak(Box::new(message_handler));
+		// acquire a permit before attempting to receive a message, exit if either fails
+		while let (Ok(permit),Some(message_result))=(SEM.acquire().await,messages.next().await){
+			// handle the message on a new thread (mainly to decode the model file)
+			tokio::spawn(async{
+				match message_handler.handle_message_result(message_result).await{
+					Ok(())=>println!("[Validation] Success, hooray!"),
+					Err(e)=>println!("[Validation] There was an error, oopsie! {e}"),
+				}
+				// explicitly call drop to make the move semantics and permit release more obvious
+				core::mem::drop(permit);
+			});
+		}
+	};
+
+	// race sigkill and main loop termination and then die
+	tokio::select!{
+		_=sig_term.recv()=>(),
+		_=main_loop=>(),
+	};
 
 	Ok(())
 }

validation/src/message_handler.rs

@@ -0,0 +1,48 @@
+#[allow(dead_code)]
+#[derive(Debug)]
+pub enum HandleMessageError{
+	Messages(async_nats::jetstream::consumer::pull::MessagesError),
+	DoubleAck(async_nats::Error),
+	UnknownSubject(String),
+	PublishNew(crate::publish_new::PublishError),
+	PublishFix(crate::publish_fix::PublishError),
+	Validation(crate::validator::ValidateError),
+}
+impl std::fmt::Display for HandleMessageError{
+	fn fmt(&self,f:&mut std::fmt::Formatter<'_>)->std::fmt::Result{
+		write!(f,"{self:?}")
+	}
+}
+impl std::error::Error for HandleMessageError{}
+
+pub type MessageResult=Result<async_nats::jetstream::Message,async_nats::jetstream::consumer::pull::MessagesError>;
+
+pub struct MessageHandler{
+	publish_new:crate::publish_new::Publisher,
+	publish_fix:crate::publish_fix::Publisher,
+	validator:crate::validator::Validator,
+}
+
+impl MessageHandler{
+	pub fn new(
+		cookie_context:rbx_asset::cookie::CookieContext,
+		api:api::Context,
+		maps_grpc:crate::MapsServiceClient,
+	)->Self{
+		Self{
+			publish_new:crate::publish_new::Publisher::new(cookie_context.clone(),api.clone(),maps_grpc),
+			publish_fix:crate::publish_fix::Publisher::new(cookie_context.clone(),api.clone()),
+			validator:crate::validator::Validator::new(cookie_context,api),
+		}
+	}
+	pub async fn handle_message_result(&self,message_result:MessageResult)->Result<(),HandleMessageError>{
+		let message=message_result.map_err(HandleMessageError::Messages)?;
+		message.double_ack().await.map_err(HandleMessageError::DoubleAck)?;
+		match message.subject.as_str(){
+			"maptest.submissions.publishnew"=>self.publish_new.publish(message).await.map_err(HandleMessageError::PublishNew),
+			"maptest.submissions.publishfix"=>self.publish_fix.publish(message).await.map_err(HandleMessageError::PublishFix),
+			"maptest.submissions.validate"=>self.validator.validate(message).await.map_err(HandleMessageError::Validation),
+			other=>Err(HandleMessageError::UnknownSubject(other.to_owned()))
+		}
+	}
+}

validation/src/publish_fix.rs

@@ -1,12 +1,8 @@
-use futures::StreamExt;
-use crate::types::{MessageResult,NatsStartupError};
 use crate::nats_types::PublishFixRequest;
 
 #[allow(dead_code)]
 #[derive(Debug)]
-enum PublishError{
+pub enum PublishError{
-	Messages(async_nats::jetstream::consumer::pull::MessagesError),
-	DoubleAck(async_nats::Error),
 	Get(rbx_asset::cookie::GetError),
 	Json(serde_json::Error),
 	Upload(rbx_asset::cookie::UploadError),
@@ -20,43 +16,21 @@ impl std::fmt::Display for PublishError{
 impl std::error::Error for PublishError{}
 
 pub struct Publisher{
-	messages:async_nats::jetstream::consumer::pull::Stream,
 	roblox_cookie:rbx_asset::cookie::CookieContext,
 	api:api::Context,
 }
 impl Publisher{
-	pub async fn new(
+	pub const fn new(
-		stream:async_nats::jetstream::stream::Stream,
 		roblox_cookie:rbx_asset::cookie::CookieContext,
 		api:api::Context,
-	)->Result<Self,NatsStartupError>{
+	)->Self{
-		Ok(Self{
+		Self{
-			messages:stream.get_or_create_consumer("publish_fix",async_nats::jetstream::consumer::pull::Config{
-				name:Some("publish_fix".to_owned()),
-				durable_name:Some("publish_fix".to_owned()),
-				filter_subject:"maptest.submissions.publish.fix".to_owned(),
-				..Default::default()
-			}).await.map_err(NatsStartupError::Consumer)?
-			.messages().await.map_err(NatsStartupError::Stream)?,
 			roblox_cookie,
 			api,
-		})
-	}
-	pub async fn run(mut self){
-		while let Some(message_result)=self.messages.next().await{
-			self.publish_supress_error(message_result).await
 		}
 	}
-	async fn publish_supress_error(&self,message_result:MessageResult){
+	pub async fn publish(&self,message:async_nats::jetstream::Message)->Result<(),PublishError>{
-		match self.publish(message_result).await{
+		println!("publish_fix {:?}",message.message.payload);
-			Ok(())=>println!("[PublishFix] Published, hooray!"),
-			Err(e)=>println!("[PublishFix] There was an error, oopsie! {e}"),
-		}
-	}
-	async fn publish(&self,message_result:MessageResult)->Result<(),PublishError>{
-		println!("publish_fix {:?}",message_result);
-		let message=message_result.map_err(PublishError::Messages)?;
-		message.double_ack().await.map_err(PublishError::DoubleAck)?;
 		// decode json
 		let publish_info:PublishFixRequest=serde_json::from_slice(&message.payload).map_err(PublishError::Json)?;
 

validation/src/publish_new.rs

@@ -1,12 +1,8 @@
-use futures::StreamExt;
-use crate::types::{MessageResult,NatsStartupError};
 use crate::nats_types::PublishNewRequest;
 
 #[allow(dead_code)]
 #[derive(Debug)]
-enum PublishError{
+pub enum PublishError{
-	Messages(async_nats::jetstream::consumer::pull::MessagesError),
-	DoubleAck(async_nats::Error),
 	Get(rbx_asset::cookie::GetError),
 	Json(serde_json::Error),
 	Create(rbx_asset::cookie::CreateError),
@@ -22,46 +18,24 @@ impl std::fmt::Display for PublishError{
 impl std::error::Error for PublishError{}
 
 pub struct Publisher{
-	messages:async_nats::jetstream::consumer::pull::Stream,
 	roblox_cookie:rbx_asset::cookie::CookieContext,
 	api:api::Context,
-	maps_grpc:crate::types::MapsServiceClient,
+	maps_grpc:crate::MapsServiceClient,
 }
 impl Publisher{
-	pub async fn new(
+	pub const fn new(
-		stream:async_nats::jetstream::stream::Stream,
 		roblox_cookie:rbx_asset::cookie::CookieContext,
 		api:api::Context,
-		maps_grpc:crate::types::MapsServiceClient,
+		maps_grpc:crate::MapsServiceClient,
-	)->Result<Self,NatsStartupError>{
+	)->Self{
-		Ok(Self{
+		Self{
-			messages:stream.get_or_create_consumer("publish_new",async_nats::jetstream::consumer::pull::Config{
-				name:Some("publish_new".to_owned()),
-				durable_name:Some("publish_new".to_owned()),
-				filter_subject:"maptest.submissions.publish.new".to_owned(),
-				..Default::default()
-			}).await.map_err(NatsStartupError::Consumer)?
-			.messages().await.map_err(NatsStartupError::Stream)?,
 			roblox_cookie,
 			api,
 			maps_grpc,
-		})
-	}
-	pub async fn run(mut self){
-		while let Some(message_result)=self.messages.next().await{
-			self.publish_supress_error(message_result).await
 		}
 	}
-	async fn publish_supress_error(&self,message_result:MessageResult){
+	pub async fn publish(&self,message:async_nats::jetstream::Message)->Result<(),PublishError>{
-		match self.publish(message_result).await{
+		println!("publish_new {:?}",message.message.payload);
-			Ok(())=>println!("[PublishNew] Published, hooray!"),
-			Err(e)=>println!("[PublishNew] There was an error, oopsie! {e}"),
-		}
-	}
-	async fn publish(&self,message_result:MessageResult)->Result<(),PublishError>{
-		println!("publish_new {:?}",message_result);
-		let message=message_result.map_err(PublishError::Messages)?;
-		message.double_ack().await.map_err(PublishError::DoubleAck)?;
 		// decode json
 		let publish_info:PublishNewRequest=serde_json::from_slice(&message.payload).map_err(PublishError::Json)?;
 

validation/src/types.rs

@@ -1,16 +0,0 @@
-
-// annoying mile-long types
-pub type MapsServiceClient=rust_grpc::maps::maps_service_client::MapsServiceClient<tonic::transport::channel::Channel>;
-pub type MessageResult=Result<async_nats::jetstream::Message,async_nats::jetstream::consumer::pull::MessagesError>;
-
-#[derive(Debug)]
-pub enum NatsStartupError{
-	Consumer(async_nats::jetstream::stream::ConsumerError),
-	Stream(async_nats::jetstream::consumer::StreamError),
-}
-impl std::fmt::Display for NatsStartupError{
-	fn fmt(&self,f:&mut std::fmt::Formatter<'_>)->std::fmt::Result{
-		write!(f,"{self:?}")
-	}
-}
-impl std::error::Error for NatsStartupError{}

validation/src/validator.rs

@@ -1,5 +1,5 @@
-use futures::{StreamExt,TryStreamExt};
+use futures::TryStreamExt;
-use crate::types::{MessageResult,NatsStartupError};
+
 use crate::nats_types::ValidateRequest;
 
 const SCRIPT_CONCURRENCY:usize=16;
@@ -13,11 +13,9 @@ enum Policy{
 
 #[allow(dead_code)]
 #[derive(Debug)]
-enum ValidateError{
+pub enum ValidateError{
 	Blocked,
 	NotAllowed,
-	Messages(async_nats::jetstream::consumer::pull::MessagesError),
-	DoubleAck(async_nats::Error),
 	Get(rbx_asset::cookie::GetError),
 	Json(serde_json::Error),
 	ReadDom(ReadDomError),
@@ -37,45 +35,22 @@ impl std::fmt::Display for ValidateError{
 impl std::error::Error for ValidateError{}
 
 pub struct Validator{
-	messages:async_nats::jetstream::consumer::pull::Stream,
 	roblox_cookie:rbx_asset::cookie::CookieContext,
 	api:api::Context,
 }
 
 impl Validator{
-	pub async fn new(
+	pub const fn new(
-		stream:async_nats::jetstream::stream::Stream,
 		roblox_cookie:rbx_asset::cookie::CookieContext,
 		api:api::Context,
-	)->Result<Self,NatsStartupError>{
+	)->Self{
-		Ok(Self{
+		Self{
-			messages:stream.get_or_create_consumer("validation",async_nats::jetstream::consumer::pull::Config{
-				name:Some("validation".to_owned()),
-				durable_name:Some("validation".to_owned()),
-				ack_policy:async_nats::jetstream::consumer::AckPolicy::Explicit,
-				filter_subject:"maptest.submissions.validate".to_owned(),
-				..Default::default()
-			}).await.map_err(NatsStartupError::Consumer)?
-			.messages().await.map_err(NatsStartupError::Stream)?,
 			roblox_cookie,
 			api,
-		})
-	}
-	pub async fn run(mut self){
-		while let Some(message_result)=self.messages.next().await{
-			self.validate_supress_error(message_result).await
 		}
 	}
-	async fn validate_supress_error(&self,message_result:MessageResult){
+	pub async fn validate(&self,message:async_nats::jetstream::Message)->Result<(),ValidateError>{
-		match self.validate(message_result).await{
+		println!("validate {:?}",message.message.payload);
-			Ok(())=>println!("[Validation] Validated, hooray!"),
-			Err(e)=>println!("[Validation] There was an error, oopsie! {e}"),
-		}
-	}
-	async fn validate(&self,message_result:MessageResult)->Result<(),ValidateError>{
-		println!("validate {:?}",message_result);
-		let message=message_result.map_err(ValidateError::Messages)?;
-		message.double_ack().await.map_err(ValidateError::DoubleAck)?;
 		// decode json
 		let validate_info:ValidateRequest=serde_json::from_slice(&message.payload).map_err(ValidateError::Json)?;
 
@@ -158,10 +133,10 @@ impl Validator{
 		if modified{
 			// serialize model (slow!)
 			let mut data=Vec::new();
-			rbx_binary::to_writer(&mut data,&dom,&[dom.root_ref()]).map_err(ValidateError::WriteDom)?;
+			rbx_binary::to_writer(&mut data,&dom,dom.root().children()).map_err(ValidateError::WriteDom)?;
 
 			// upload a model lol
-			let (model_id,model_version)=if let Some(model_id)=validate_info.ValidatedModelID{
+			let model_id=if let Some(model_id)=validate_info.ValidatedModelID{
 				// upload to existing id
 				let response=self.roblox_cookie.upload(rbx_asset::cookie::UploadRequest{
 					assetid:model_id,
@@ -172,7 +147,7 @@ impl Validator{
 					groupId:None,
 				},data).await.map_err(ValidateError::Upload)?;
 
-				(response.AssetId,response.AssetVersionId)
+				response.AssetId
 			}else{
 				// create new model
 				let response=self.roblox_cookie.create(rbx_asset::cookie::CreateRequest{
@@ -183,14 +158,14 @@ impl Validator{
 					groupId:None,
 				},data).await.map_err(ValidateError::Create)?;
 
-				(response.AssetId,response.AssetVersionId)
+				response.AssetId
 			};
 
 			// update the submission to use the validated model
 			self.api.update_submission_model(api::UpdateSubmissionModelRequest{
 				ID:validate_info.SubmissionID,
 				ModelID:model_id,
-				ModelVersion:model_version,
+				ModelVersion:1, //TODO
 			}).await.map_err(ValidateError::ApiUpdateSubmissionModel)?;
 		};
 
@@ -205,7 +180,7 @@ impl Validator{
 
 #[allow(dead_code)]
 #[derive(Debug)]
-enum ReadDomError{
+pub enum ReadDomError{
 	Binary(rbx_binary::DecodeError),
 	Xml(rbx_xml::DecodeError),
 	Read(std::io::Error),

web/next.config.ts

@@ -6,8 +6,8 @@ const nextConfig: NextConfig = {
     rewrites: async () => {
     	return [
      		{
-       			source: "/v1/submissions/1/status/:statustype",
+       			source: "/v1/submissions/:submissionid/status/:statustype",
-          		destination: "http://localhost:8081/v1/submissions/:statustype"
+          		destination: "http://mapsservice:8082/v1/submissions/:submissionid/status/:statustype"
        		}
      	]
     }

web/src/app/submissions/[submissionId]/_reviewButtons.tsx

@@ -1,6 +1,6 @@
 import { Button, ButtonOwnProps } from "@mui/material";
 
-type Review = "Completed" | "Submit" | "Reject" | "Revoke" | "Validate" | "Publish"
+type Review = "Completed" | "Submit" | "Reject" | "Revoke" | "Accept" | "Publish"
 type Action = "completed" | "submit" | "reject" | "revoke" | "trigger-validate" | "trigger-publish"
 interface ReviewButton {
 	name: Review,
@@ -13,7 +13,6 @@ function ReviewButtonClicked(action: Action) {
 		method: "POST",
 		headers: {
 			"Content-type": "application/json",
-			"Cookie": "session_id=c5191ddc-eee1-4010-900c-6b2c7b6780ab"
 		}
 	})
 }
@@ -25,12 +24,12 @@ function ReviewButton(props: ReviewButton) {
 export default function ReviewButtons() {
 	return (
 		<section className="review-set">
-			<ReviewButton color="error" name="Reject"    action="reject"/>
+			<ReviewButton color="info"  name="Submit"    action="submit"/>
 			<ReviewButton color="info"  name="Revoke"    action="revoke"/>
+			<ReviewButton color="info"  name="Accept"    action="trigger-validate"/>
+			<ReviewButton color="error" name="Reject"    action="reject"/>
 			<ReviewButton color="info"  name="Publish"   action="trigger-publish"/>
 			<ReviewButton color="info"  name="Completed" action="completed"/>
-			<ReviewButton color="info"  name="Submit"    action="submit"/>
-			<ReviewButton color="info"  name="Validate"  action="trigger-validate"/>
 		</section>
 	)
 }