feat(status_events): update values on startup

shard.lock

@@ -14,20 +14,32 @@ shards:
 
   ameba:
     git: https://github.com/crystal-ameba/ameba.git
-    version: 1.2.0
+    version: 1.3.0
 
   backtracer:
     git: https://github.com/sija/backtracer.cr.git
     version: 1.2.2
 
   bindata:
     git: https://github.com/spider-gazelle/bindata.git
-    version: 1.11.0
+    version: 1.11.1
+
+  connect-proxy:
+    git: https://github.com/spider-gazelle/connect-proxy.git
+    version: 2.0.0
+
+  crc16:
+    git: https://github.com/maiha/crc16.cr.git
+    version: 0.1.0
 
   cron_parser:
     git: https://github.com/kostya/cron_parser.git
     version: 0.4.0
 
+  crunits:
+    git: https://github.com/spider-gazelle/crunits.git
+    version: 1.1.1
+
   csuuid:
     git: https://github.com/wyhaines/csuuid.cr.git
     version: 1.0.0+git.commit.812ee5e1252b27ee2895ec7897094cde3e4a5dc2
@@ -44,6 +56,14 @@ shards:
     git: https://github.com/wyhaines/defined.cr.git
     version: 0.3.6
 
+  dtls:
+    git: https://github.com/spider-gazelle/crystal-dtls.git
+    version: 0.9.0
+
+  etcd:
+    git: https://github.com/place-labs/crystal-etcd.git
+    version: 1.2.5
+
   exception_page:
     git: https://github.com/crystal-loot/exception_page.git
     version: 0.3.0
@@ -64,6 +84,10 @@ shards:
     git: https://github.com/luckyframework/habitat.git
     version: 0.4.7
 
+  hound-dog:
+    git: https://github.com/place-labs/hound-dog.git
+    version: 2.9.0
+
   hot_topic:
     git: https://github.com/jgaskins/hot_topic.git
     version: 0.1.0+git.commit.c4577d949221d535f29162343bf503b578308954
@@ -72,6 +96,10 @@ shards:
     git: https://github.com/place-labs/http-params-serializable.git
     version: 0.5.0
 
+  ipaddress:
+    git: https://github.com/sija/ipaddress.cr.git
+    version: 0.2.1
+
   json-merge-patch:
     git: https://github.com/caspiano/json-merge-patch.git
     version: 0.1.1
@@ -96,13 +124,17 @@ shards:
     git: https://github.com/spider-gazelle/crystal-mqtt.git
     version: 1.2.1
 
+  murmur3:
+    git: https://github.com/aca-labs/murmur3.git
+    version: 0.1.1+git.commit.7cbe25c0ca8d052c9d98c377c824dcb0e038c790
+
   nbchannel:
     git: https://github.com/wyhaines/nbchannel.cr.git
     version: 0.1.0+git.commit.ea6517a837270361695244d14c4b03d84660f113
 
   neuroplastic:
     git: https://github.com/spider-gazelle/neuroplastic.git
-    version: 1.12.1
+    version: 1.12.2
 
   open_api:
     git: https://github.com/elbywan/open_api.cr.git
@@ -122,7 +154,7 @@ shards:
 
   opentelemetry-instrumentation:
     git: https://github.com/wyhaines/opentelemetry-instrumentation.cr.git
-    version: 0.5.0+git.commit.7d83b9a53c9540fbc159f06eeed5d4bdad5c4377
+    version: 0.5.1+git.commit.30000cdca2d17db6f42db2eafc49f109f57a5afc
 
   opentelemetry-sdk:
     git: https://github.com/wyhaines/opentelemetry-sdk.cr.git
@@ -136,13 +168,21 @@ shards:
     git: https://github.com/wyhaines/parsedate.cr.git
     version: 0.1.2
 
+  placeos-core-client:
+    git: https://github.com/placeos/core-client.git
+    version: 0.5.2
+
+  placeos-driver:
+    git: https://github.com/placeos/driver.git
+    version: 6.6.0
+
   placeos-log-backend:
     git: https://github.com/place-labs/log-backend.git
     version: 0.11.0
 
   placeos-models:
     git: https://github.com/placeos/models.git
-    version: 8.12.1
+    version: 8.12.2
 
   placeos-resource:
     git: https://github.com/place-labs/resource.git
@@ -164,6 +204,18 @@ shards:
     git: https://github.com/stefanwille/crystal-redis.git
     version: 2.8.3
 
+  redis-cluster:
+    git: https://github.com/caspiano/redis-cluster.cr.git
+    version: 0.8.5
+
+  rendezvous-hash:
+    git: https://github.com/caspiano/rendezvous-hash.git
+    version: 0.3.1
+
+  responsible:
+    git: https://github.com/place-labs/responsible.git
+    version: 1.2.3
+
   rethinkdb:
     git: https://github.com/kingsleyh/crystal-rethinkdb.git
     version: 0.3.1
@@ -192,6 +244,10 @@ shards:
     git: https://github.com/wyhaines/splay_tree_map.cr.git
     version: 0.2.2
 
+  ssh2:
+    git: https://github.com/spider-gazelle/ssh2.cr.git
+    version: 1.5.3
+
   tasker:
     git: https://github.com/spider-gazelle/tasker.git
     version: 2.0.7
@@ -208,3 +264,7 @@ shards:
     git: https://github.com/wyhaines/tracer.cr.git
     version: 0.3.1
 
+  ulid:
+    git: https://github.com/place-labs/ulid.git
+    version: 0.1.2
+

shard.yml

@@ -9,6 +9,10 @@ targets:
     main: src/app.cr
 
 dependencies:
+  placeos-driver:
+    github: placeos/driver
+    version: ~> 6.3
+
   action-controller:
     github: spider-gazelle/action-controller
     version: ~> 5.3

src/source/status_events.cr

@@ -1,6 +1,7 @@
 require "redis"
 require "simple_retry"
 
+require "placeos-driver/storage"
 require "./mappings"
 require "./publishing/publisher"
 require "./publishing/publisher_manager"
@@ -20,8 +21,11 @@ module PlaceOS::Source
     def initialize(@mappings : Mappings, @publisher_managers : Array(PublisherManager))
     end
 
+    @update_mutex : Mutex = Mutex.new
+
     def start
       self.stopped = false
+      spawn(same_thread: true) { @update_mutex.synchronize { update_values } }
 
       SimpleRetry.try_to(
         base_interval: 500.milliseconds,
@@ -40,14 +44,38 @@ module PlaceOS::Source
 
     def stop
       self.stopped = true
+
       return unless @redis
       begin
         redis.punsubscribe(STATUS_CHANNEL_PATTERN)
       rescue
       end
+
       redis.close
     end
 
+    def update_values
+      mods_mapped = 0_u64
+      status_updated = 0_u64
+      pattern = "initial_sync"
+      PlaceOS::Model::Module.all.in_groups_of(64, reuse: true) do |modules|
+        modules.each do |mod|
+          break unless mod
+          mods_mapped += 1_u64
+          store = PlaceOS::Driver::RedisStorage.new(mod.id.to_s)
+          store.each do |key, value|
+            status_updated += 1_u64
+            handle_pevent(pattern: pattern, channel: key, payload: value)
+          end
+        end
+      end
+      Log.info { {
+        message: "initial status sync complete",
+        modules: mods_mapped.to_s,
+        values:  status_updated.to_s,
+      } }
+    end
+
     protected def handle_pevent(pattern : String, channel : String, payload : String)
       module_id, status = StatusEvents.parse_channel(channel)
       events = mappings.status_events?(module_id, status)