bring in devenv and move more processing into the peer

.devenv.flake.nix

···
+{
+  inputs =
+    let
+      version = "1.10.0";
+system = "x86_64-linux";
+devenv_root = "/home/jonathan/code/skypod-node";
+devenv_dotfile = "/home/jonathan/code/skypod-node/.devenv";
+devenv_dotfile_path = ./.devenv;
+devenv_tmpdir = "/run/user/1000";
+devenv_runtime = "/run/user/1000/devenv-d48b567";
+devenv_istesting = false;
+devenv_direnvrc_latest_version = 1;
+container_name = null;
+active_profiles = [ ];
+hostname = "buzz";
+username = "jonathan";
+git_root = "/home/jonathan/code/skypod-node";
+
+        in {
+        git-hooks.url = "github:cachix/git-hooks.nix";
+      git-hooks.inputs.nixpkgs.follows = "nixpkgs";
+      pre-commit-hooks.follows = "git-hooks";
+      nixpkgs.url = "github:cachix/devenv-nixpkgs/rolling";
+      devenv.url = "github:cachix/devenv?dir=src/modules";
+      } // (if builtins.pathExists (devenv_dotfile_path + "/flake.json")
+      then builtins.fromJSON (builtins.readFile (devenv_dotfile_path +  "/flake.json"))
+      else { });
+
+      outputs = { nixpkgs, ... }@inputs:
+        let
+          version = "1.10.0";
+system = "x86_64-linux";
+devenv_root = "/home/jonathan/code/skypod-node";
+devenv_dotfile = "/home/jonathan/code/skypod-node/.devenv";
+devenv_dotfile_path = ./.devenv;
+devenv_tmpdir = "/run/user/1000";
+devenv_runtime = "/run/user/1000/devenv-d48b567";
+devenv_istesting = false;
+devenv_direnvrc_latest_version = 1;
+container_name = null;
+active_profiles = [ ];
+hostname = "buzz";
+username = "jonathan";
+git_root = "/home/jonathan/code/skypod-node";
+
+            devenv =
+            if builtins.pathExists (devenv_dotfile_path + "/devenv.json")
+            then builtins.fromJSON (builtins.readFile (devenv_dotfile_path + "/devenv.json"))
+            else { };
+
+          systems = [ "x86_64-linux" "aarch64-linux" "x86_64-darwin" "aarch64-darwin" ];
+
+          # Function to create devenv configuration for a specific system with profiles support
+          mkDevenvForSystem = targetSystem:
+            let
+              getOverlays = inputName: inputAttrs:
+                map
+                  (overlay:
+                    let
+                      input = inputs.${inputName} or (throw "No such input `${inputName}` while trying to configure overlays.");
+                    in
+                      input.overlays.${overlay} or (throw "Input `${inputName}` has no overlay called `${overlay}`. Supported overlays: ${nixpkgs.lib.concatStringsSep ", " (builtins.attrNames input.overlays)}"))
+                  inputAttrs.overlays or [ ];
+              overlays = nixpkgs.lib.flatten (nixpkgs.lib.mapAttrsToList getOverlays (devenv.inputs or { }));
+              permittedUnfreePackages = devenv.nixpkgs.per-platform."${targetSystem}".permittedUnfreePackages or devenv.nixpkgs.permittedUnfreePackages or [ ];
+              pkgs = import nixpkgs {
+                system = targetSystem;
+                config = {
+                  allowUnfree = devenv.nixpkgs.per-platform."${targetSystem}".allowUnfree or devenv.nixpkgs.allowUnfree or devenv.allowUnfree or false;
+                  allowBroken = devenv.nixpkgs.per-platform."${targetSystem}".allowBroken or devenv.nixpkgs.allowBroken or devenv.allowBroken or false;
+                  cudaSupport = devenv.nixpkgs.per-platform."${targetSystem}".cudaSupport or devenv.nixpkgs.cudaSupport or false;
+                  cudaCapabilities = devenv.nixpkgs.per-platform."${targetSystem}".cudaCapabilities or devenv.nixpkgs.cudaCapabilities or [ ];
+                  permittedInsecurePackages = devenv.nixpkgs.per-platform."${targetSystem}".permittedInsecurePackages or devenv.nixpkgs.permittedInsecurePackages or devenv.permittedInsecurePackages or [ ];
+                  allowUnfreePredicate = if (permittedUnfreePackages != [ ]) then (pkg: builtins.elem (nixpkgs.lib.getName pkg) permittedUnfreePackages) else (_: false);
+                };
+                inherit overlays;
+              };
+              lib = pkgs.lib;
+              importModule = path:
+                if lib.hasPrefix "./" path
+                then if lib.hasSuffix ".nix" path
+                then ./. + (builtins.substring 1 255 path)
+                else ./. + (builtins.substring 1 255 path) + "/devenv.nix"
+                else if lib.hasPrefix "../" path
+                then
+                # For parent directory paths, concatenate with /.
+                # ./. refers to the directory containing this file (project root)
+                # So ./. + "/../shared" = <project-root>/../shared
+                  if lib.hasSuffix ".nix" path
+                  then ./. + "/${path}"
+                  else ./. + "/${path}/devenv.nix"
+                else
+                  let
+                    paths = lib.splitString "/" path;
+                    name = builtins.head paths;
+                    input = inputs.${name} or (throw "Unknown input ${name}");
+                    subpath = "/${lib.concatStringsSep "/" (builtins.tail paths)}";
+                    devenvpath = "${input}" + subpath;
+                    devenvdefaultpath = devenvpath + "/devenv.nix";
+                  in
+                  if lib.hasSuffix ".nix" devenvpath
+                  then devenvpath
+                  else if builtins.pathExists devenvdefaultpath
+                  then devenvdefaultpath
+                  else throw (devenvdefaultpath + " file does not exist for input ${name}.");
+
+              # Phase 1: Base evaluation to extract profile definitions
+              baseProject = pkgs.lib.evalModules {
+                specialArgs = inputs // { inherit inputs; };
+                modules = [
+                  ({ config, ... }: {
+                    _module.args.pkgs = pkgs.appendOverlays (config.overlays or [ ]);
+                  })
+                  (inputs.devenv.modules + /top-level.nix)
+                  ({ options, ... }: {
+                    config.devenv = lib.mkMerge [
+                      {
+                        cliVersion = version;
+                        root = devenv_root;
+                        dotfile = devenv_dotfile;
+                      }
+                      (pkgs.lib.optionalAttrs (builtins.hasAttr "tmpdir" options.devenv) {
+                        tmpdir = devenv_tmpdir;
+                      })
+                      (pkgs.lib.optionalAttrs (builtins.hasAttr "isTesting" options.devenv) {
+                        isTesting = devenv_istesting;
+                      })
+                      (pkgs.lib.optionalAttrs (builtins.hasAttr "runtime" options.devenv) {
+                        runtime = devenv_runtime;
+                      })
+                      (pkgs.lib.optionalAttrs (builtins.hasAttr "direnvrcLatestVersion" options.devenv) {
+                        direnvrcLatestVersion = devenv_direnvrc_latest_version;
+                      })
+                    ];
+                  })
+                  ({ options, ... }: {
+                    config = lib.mkMerge [
+                      (pkgs.lib.optionalAttrs (builtins.hasAttr "git" options) {
+                        git.root = git_root;
+                      })
+                    ];
+                  })
+                  (pkgs.lib.optionalAttrs (container_name != null) {
+                    container.isBuilding = pkgs.lib.mkForce true;
+                    containers.${container_name}.isBuilding = true;
+                  })
+                ] ++ (map importModule (devenv.imports or [ ])) ++ [
+                  (if builtins.pathExists ./devenv.nix then ./devenv.nix else { })
+                  (devenv.devenv or { })
+                  (if builtins.pathExists ./devenv.local.nix then ./devenv.local.nix else { })
+                  (if builtins.pathExists (devenv_dotfile_path + "/cli-options.nix") then import (devenv_dotfile_path + "/cli-options.nix") else { })
+                ];
+              };
+
+              # Phase 2: Extract and apply profiles using extendModules with priority overrides
+              project =
+                let
+                  # Build ordered list of profile names: hostname -> user -> manual
+                  manualProfiles = active_profiles;
+                  currentHostname = hostname;
+                  currentUsername = username;
+                  hostnameProfiles = lib.optional (currentHostname != "" && builtins.hasAttr currentHostname (baseProject.config.profiles.hostname or { })) "hostname.${currentHostname}";
+                  userProfiles = lib.optional (currentUsername != "" && builtins.hasAttr currentUsername (baseProject.config.profiles.user or { })) "user.${currentUsername}";
+
+                  # Ordered list of profiles to activate
+                  orderedProfiles = hostnameProfiles ++ userProfiles ++ manualProfiles;
+
+                  # Resolve profile extends with cycle detection
+                  resolveProfileExtends = profileName: visited:
+                    if builtins.elem profileName visited then
+                      throw "Circular dependency detected in profile extends: ${lib.concatStringsSep " -> " visited} -> ${profileName}"
+                    else
+                      let
+                        profile = getProfileConfig profileName;
+                        extends = profile.extends or [ ];
+                        newVisited = visited ++ [ profileName ];
+                        extendedProfiles = lib.flatten (map (name: resolveProfileExtends name newVisited) extends);
+                      in
+                      extendedProfiles ++ [ profileName ];
+
+                  # Get profile configuration by name from baseProject
+                  getProfileConfig = profileName:
+                    if lib.hasPrefix "hostname." profileName then
+                      let name = lib.removePrefix "hostname." profileName;
+                      in baseProject.config.profiles.hostname.${name}
+                    else if lib.hasPrefix "user." profileName then
+                      let name = lib.removePrefix "user." profileName;
+                      in baseProject.config.profiles.user.${name}
+                    else
+                      let
+                        availableProfiles = builtins.attrNames (baseProject.config.profiles or { });
+                        hostnameProfiles = map (n: "hostname.${n}") (builtins.attrNames (baseProject.config.profiles.hostname or { }));
+                        userProfiles = map (n: "user.${n}") (builtins.attrNames (baseProject.config.profiles.user or { }));
+                        allAvailableProfiles = availableProfiles ++ hostnameProfiles ++ userProfiles;
+                      in
+                        baseProject.config.profiles.${profileName} or (throw "Profile '${profileName}' not found. Available profiles: ${lib.concatStringsSep ", " allAvailableProfiles}");
+
+                  # Fold over ordered profiles to build final list with extends
+                  expandedProfiles = lib.foldl'
+                    (acc: profileName:
+                      let
+                        allProfileNames = resolveProfileExtends profileName [ ];
+                      in
+                      acc ++ allProfileNames
+                    ) [ ]
+                    orderedProfiles;
+
+                  # Map over expanded profiles and apply priorities
+                  allPrioritizedModules = lib.imap0
+                    (index: profileName:
+                      let
+                        # Decrement priority for each profile (lower = higher precedence)
+                        # Start with the next lowest priority after the default priority for values (100)
+                        profilePriority = (lib.modules.defaultOverridePriority - 1) - index;
+                        profileConfig = getProfileConfig profileName;
+
+                        # Check if an option type needs explicit override to resolve conflicts
+                        # Only apply overrides to LEAF values (scalars), not collection types that can merge
+                        typeNeedsOverride = type:
+                          if type == null then false
+                          else
+                            let
+                              typeName = type.name or type._type or "";
+
+                              # True leaf types that need priority resolution when they conflict
+                              isLeafType = builtins.elem typeName [
+                                "str"
+                                "int"
+                                "bool"
+                                "enum"
+                                "path"
+                                "package"
+                                "float"
+                                "anything"
+                              ];
+                            in
+                            if isLeafType then true
+                            else if typeName == "nullOr" then
+                            # For nullOr, check the wrapped type recursively
+                              let
+                                innerType = type.elemType or
+                                  (if type ? nestedTypes && type.nestedTypes ? elemType
+                                  then type.nestedTypes.elemType
+                                  else null);
+                              in
+                              if innerType != null then typeNeedsOverride innerType else false
+                            else
+                            # Everything else (collections, submodules, etc.) should merge naturally
+                              false;
+
+                        # Check if a config path needs explicit override
+                        pathNeedsOverride = optionPath:
+                          let
+                            # Try direct option first
+                            directOption = lib.attrByPath optionPath null baseProject.options;
+                          in
+                          if directOption != null && lib.isOption directOption then
+                            typeNeedsOverride directOption.type
+                          else if optionPath != [ ] then
+                          # Check parent for freeform type
+                            let
+                              parentPath = lib.init optionPath;
+                              parentOption = lib.attrByPath parentPath null baseProject.options;
+                            in
+                            if parentOption != null && lib.isOption parentOption then
+                              let
+                                # Look for freeform type:
+                                # 1. Standard location: type.freeformType (primary)
+                                # 2. Nested location: type.nestedTypes.freeformType (evaluated form)
+                                freeformType = parentOption.type.freeformType or
+                                  parentOption.type.nestedTypes.freeformType or
+                                    null;
+                                elementType =
+                                  if freeformType ? elemType then freeformType.elemType
+                                  else if freeformType ? nestedTypes && freeformType.nestedTypes ? elemType then freeformType.nestedTypes.elemType
+                                  else freeformType;
+                              in
+                              typeNeedsOverride elementType
+                            else false
+                          else false;
+
+                        # Support overriding both plain attrset modules and functions
+                        applyModuleOverride = config:
+                          if builtins.isFunction config
+                          then
+                            let
+                              wrapper = args: applyOverrideRecursive (config args) [ ];
+                            in
+                            lib.mirrorFunctionArgs config wrapper
+                          else applyOverrideRecursive config [ ];
+
+                        # Apply overrides recursively based on option types
+                        applyOverrideRecursive = config: optionPath:
+                          if lib.isAttrs config && config ? _type then
+                            config  # Don't touch values with existing type metadata
+                          else if lib.isAttrs config then
+                            lib.mapAttrs (name: value: applyOverrideRecursive value (optionPath ++ [ name ])) config
+                          else if pathNeedsOverride optionPath then
+                            lib.mkOverride profilePriority config
+                          else
+                            config;
+
+                        # Apply priority overrides recursively to the deferredModule imports structure
+                        prioritizedConfig = (
+                          profileConfig.module // {
+                            imports = lib.map
+                              (importItem:
+                                importItem // {
+                                  imports = lib.map
+                                    (nestedImport:
+                                      applyModuleOverride nestedImport
+                                    )
+                                    (importItem.imports or [ ]);
+                                }
+                              )
+                              (profileConfig.module.imports or [ ]);
+                          }
+                        );
+                      in
+                      prioritizedConfig
+                    )
+                    expandedProfiles;
+                in
+                if allPrioritizedModules == [ ]
+                then baseProject
+                else baseProject.extendModules { modules = allPrioritizedModules; };
+
+              config = project.config;
+
+              options = pkgs.nixosOptionsDoc {
+                options = builtins.removeAttrs project.options [ "_module" ];
+                warningsAreErrors = false;
+                # Unpack Nix types, e.g. literalExpression, mDoc.
+                transformOptions =
+                  let isDocType = v: builtins.elem v [ "literalDocBook" "literalExpression" "literalMD" "mdDoc" ];
+                  in lib.attrsets.mapAttrs (_: v:
+                    if v ? _type && isDocType v._type then
+                      v.text
+                    else if v ? _type && v._type == "derivation" then
+                      v.name
+                    else
+                      v
+                  );
+              };
+
+              # Recursively search for outputs in the config.
+              # This is used when not building a specific output by attrpath.
+              build = options: config:
+                lib.concatMapAttrs
+                  (name: option:
+                    if lib.isOption option then
+                      let typeName = option.type.name or "";
+                      in
+                      if builtins.elem typeName [ "output" "outputOf" ] then
+                        { ${name} = config.${name}; }
+                      else { }
+                    else if builtins.isAttrs option && !lib.isDerivation option then
+                      let v = build option config.${name};
+                      in if v != { } then {
+                        ${name} = v;
+                      } else { }
+                    else { }
+                  )
+                  options;
+            in
+            {
+              inherit config options build project;
+              shell = config.shell;
+              packages = {
+                optionsJSON = options.optionsJSON;
+                # deprecated
+                inherit (config) info procfileScript procfileEnv procfile;
+                ci = config.ciDerivation;
+              };
+            };
+
+          # Generate per-system devenv configurations
+          perSystem = nixpkgs.lib.genAttrs systems mkDevenvForSystem;
+
+          # Default devenv for the current system
+          currentSystemDevenv = perSystem.${system};
+        in
+        {
+          devShell = nixpkgs.lib.genAttrs systems (s: perSystem.${s}.shell);
+          packages = nixpkgs.lib.genAttrs systems (s: perSystem.${s}.packages);
+
+          # Per-system devenv configurations
+          devenv = {
+            # Default devenv for the current system
+            inherit (currentSystemDevenv) config options build shell packages project;
+            # Per-system devenv configurations
+            inherit perSystem;
+          };
+
+          # Legacy build output
+          build = currentSystemDevenv.build currentSystemDevenv.options currentSystemDevenv.config;
+        };
+      }

.devenv/devenv.json

···
+{"inputs":{"nixpkgs":{"url":"github:cachix/devenv-nixpkgs/rolling"}}}

.devenv/flake.json

···
+{"nixpkgs":{"url":"github:cachix/devenv-nixpkgs/rolling"}}

.devenv/gc/shell

···
+shell-5-link

.devenv/gc/shell-5-link

···
+/nix/store/nsxv5ns3yk1sfbzvdgr75z2m0ylg8njb-devenv-shell-env

.devenv/input-paths.txt

···
+/home/jonathan/code/skypod-node/.devenv/flake.json
+/home/jonathan/code/skypod-node/.devenv.flake.nix
+/home/jonathan/code/skypod-node/.env
+/home/jonathan/code/skypod-node/devenv.local.nix
+/home/jonathan/code/skypod-node/devenv.lock
+/home/jonathan/code/skypod-node/devenv.nix
+/home/jonathan/code/skypod-node/devenv.yaml

.devenv/load-exports

···
+

.devenv/profile

···
+/nix/store/wmh3hwran6m1ln7n4ci6794xyrrgcs6w-devenv-profile

.devenv/run

···
+/run/user/1000/devenv-d48b567

src/client/realm/service-connection-peer.ts

···
 import * as protocol from '#common/protocol'
 import {IdentID} from '#common/protocol'
 
+import {BlockingQueue} from '#common/async/blocking-queue.js'
+import {sleep} from '#common/async/sleep.js'
+import z from 'zod/v4'
 import {RealmSyncManager} from './service-connection-sync'
 
+const realmRtcAutohandleSchema = z.union([
+  protocol.realmRtcPingRequestSchema,
+  protocol.realmRtcPongResponseSchema,
+])
+
 /** a single webrtc peer connection within a realm */
 export class RealmPeer extends SimplePeer {
   initiator: boolean
   identid: IdentID
 
+  #seq: number
   #sync: RealmSyncManager
-
-  #pingSeq: number
-  #pingAbort: AbortController
-  #pingTimer: ReturnType<typeof setInterval>
+  #queue: BlockingQueue<unknown>
+  #abort: AbortController
 
   constructor(sync: RealmSyncManager, identid: IdentID, initiator: boolean) {
     super({
···
     this.initiator = initiator
     this.identid = identid
 
+    this.#seq = 0
     this.#sync = sync
+    this.#queue = new BlockingQueue()
+    this.#abort = new AbortController()
 
-    this.#pingSeq = 0
-    this.#pingAbort = new AbortController()
-    this.#pingAbort.signal.addEventListener('abort', this.#pingCancel)
-    this.#pingTimer = setInterval(this.#pingSync, 30_000)
+    // we start the loops and more event handlers in the connect handler
 
-    this.on('connect', this.#handlePeerConnect.bind(this))
-    this.on('close', this.#handlePeerClose.bind(this))
-    this.on('error', this.#handlePeerError.bind(this))
+    this.on('connect', this.#connected)
+    this.on('close', this.#closed)
+    this.on('error', this.#errored)
 
-    if (initiator) {
-      this.on('connect', this.#pingSync)
-    }
+    this.#abort.signal.addEventListener('abort', this.#aborted)
   }
 
   destroy(error?: Error) {
-    this.#pingAbort.abort(error)
+    this.#abort.abort(error)
+
     super.destroy(error)
   }
 
   sendJson<T extends unknown>(data: T) {
     console.debug('sending:', this.identid, data)
+
     this.send(JSON.stringify(data))
   }
 
-  async ping(signal?: AbortSignal) {
-    signal?.throwIfAborted()
-    if (!this.connected) return
+  #dispatch(type: string, detail?: object) {
+    this.emit(type, new CustomEvent(type, {bubbles: true, detail}))
+  }
 
-    const peerSync = this.identid === this.#sync.chooseSyncPeer(this.#sync.knownPeers)
-    const peerClocks = await this.#sync.buildSyncState()
+  #connected = () => {
+    console.debug(`connected to ${this.identid}`)
 
-    console.log('sending ping to: ', this.identid, peerClocks, peerSync)
-    this.sendJson<protocol.RealmRtcPingRequest>({
-      typ: 'req',
-      msg: 'realm.rtc.ping',
-      seq: this.#pingSeq++,
-      dat: {
-        peerClocks,
-        peerSync,
-      },
+    // start loops, then start listening
+    // loops go first so we don't have messages without handling
+
+    this.#pingLoop().catch((err: unknown) => {
+      this.#errored(err as Error)
+    })
+    this.#receiveLoop().catch((err: unknown) => {
+      this.#errored(err as Error)
     })
+
+    this.on('data', this.#receive)
+
+    // announce
+    this.#dispatch('peeropen', {identid: this.identid})
   }
 
-  #pingSync = () => {
-    this.ping(this.#pingAbort.signal).catch((err: unknown) => {
-      console.error(err)
-    })
+  #aborted = () => {
+    console.log('peer connected aborted...', this.identid)
+
+    // stop listeners; loops were stopped by the abort signal
+    this.off('data', this.#receive)
+
+    // announce
+    this.#dispatch('peerabort', {identid: this.identid})
+  }
+
+  #closed = () => {
+    console.debug(`disconnected from ${this.identid}`)
+
+    this.#abort.abort()
+    this.#dispatch('peerclose', {identid: this.identid})
+  }
+
+  #errored = (err: Error) => {
+    console.error(`Error with peer ${this.identid}:`, err)
+
+    this.#dispatch('peererror', {identid: this.identid, error: err})
+  }
+
+  /// receives go into a queue
+
+  #receive = (chunk: string | Uint8Array) => {
+    const string = typeof chunk === 'string' ? chunk : new TextDecoder().decode(chunk)
+    const data: unknown = JSON.parse(string)
+
+    const messages: unknown[] = Array.isArray(data) ? data : [data]
+    this.#queue.enqueue(...messages)
   }
 
-  #pingCancel = () => {
-    console.log('shutting down ping...', this.identid)
+  #receiveLoop = async () => {
+    this.#abort.signal.throwIfAborted()
+    console.debug('receiver loop running for:', this.identid)
+
+    while (!this.#abort.signal.aborted) {
+      const message = await this.#queue.dequeue(this.#abort.signal)
+      const parsed = await realmRtcAutohandleSchema.safeParseAsync(message)
+
+      if (!parsed.success) {
+        this.#dispatch('peerdata', {identid: this.identid, data: message})
+        continue
+      }
+
+      switch (parsed.data.msg) {
+        case 'realm.rtc.ping':
+          await this.#receivePing(parsed.data)
+          continue
 
-    clearInterval(this.#pingTimer)
-    this.#pingAbort.signal.removeEventListener('abort', this.#pingCancel)
+        case 'realm.rtc.pong':
+          await this.#receivePong(parsed.data)
+          continue
+      }
+    }
   }
 
-  // when a peer sends a ping
-  incomingPing = async (ping: protocol.RealmRtcPingRequest) => {
+  #receivePing = async (ping: protocol.RealmRtcPingRequest) => {
     const peerClocks = await this.#sync.buildSyncState()
 
     // reply to pings with pongs
···
     if (ping.dat.peerSync) {
       const actions = await this.#sync.fetchSyncDelta(ping.dat.peerClocks)
       if (actions.length) {
-        console.debug('ping request is sending data:', actions)
         this.sendJson(actions.map((a) => a.action))
       }
     }
+
+    this.#dispatch('peerping', {identid: this.identid})
   }
 
   // when a peer responds to a ping (lazy sync)
-  incomingPong = async (pong: protocol.RealmRtcPongResponse) => {
+  #receivePong = async (pong: protocol.RealmRtcPongResponse) => {
     const actions = await this.#sync.fetchSyncDelta(pong.dat.peerClocks)
     if (actions.length) {
-      console.debug('pong is sending data:', actions)
       this.sendJson(actions.map((a) => a.action))
     }
   }
 
-  #dispatchCustomEvent(type: string, detail?: object) {
-    this.emit(type, new CustomEvent(type, {bubbles: true, detail}))
-  }
+  /// ping/pong
 
-  #handlePeerConnect = () => {
-    console.debug(`connected to ${this.identid}`)
-    this.on('data', this.#handlePeerData.bind(this))
+  #pingLoop = async () => {
+    this.#abort.signal.throwIfAborted()
+    console.debug('receiver loop running for:', this.identid)
 
-    this.#dispatchCustomEvent('peeropen', {identid: this.identid})
-  }
-
-  #handlePeerClose = () => {
-    console.debug(`disconnected from ${this.identid}`)
-    this.off('data', this.#handlePeerData.bind(this))
+    if (this.initiator) {
+      await this.#ping()
+    }
 
-    this.#dispatchCustomEvent('peerclose', {identid: this.identid})
+    while (!this.#abort.signal.aborted) {
+      await sleep(30_000, this.#abort.signal)
+      await this.#ping()
+    }
   }
 
-  #handlePeerError = (err: Error) => {
-    console.error(`Error with peer ${this.identid}:`, err)
-
-    this.#dispatchCustomEvent('peererror', {identid: this.identid, error: err})
-  }
+  #ping = async () => {
+    this.#abort.signal.throwIfAborted()
+    if (!this.connected) return
 
-  #handlePeerData = (chunk: string | Uint8Array) => {
-    try {
-      const string = typeof chunk === 'string' ? chunk : new TextDecoder().decode(chunk)
-      const data: unknown = JSON.parse(string)
+    const peerSync = this.identid === this.#sync.chooseSyncPeer(this.#sync.knownPeers)
+    const peerClocks = await this.#sync.buildSyncState()
 
-      const messages: unknown[] = Array.isArray(data) ? data : [data]
-      for (const message of messages) {
-        this.#dispatchCustomEvent('peermessage', {peer: this, message})
-      }
-    } catch (er) {
-      console.error(er)
-      throw er
-    }
+    console.log('sending ping to: ', this.identid, peerClocks, peerSync)
+    this.sendJson<protocol.RealmRtcPingRequest>({
+      typ: 'req',
+      msg: 'realm.rtc.ping',
+      seq: this.#seq++,
+      dat: {
+        peerClocks,
+        peerSync,
+      },
+    })
   }
 }

src/client/realm/service-connection.ts

···
 import {z} from 'zod/v4'
 
 import {timeoutSignal} from '#common/async/aborts.js'
-import {BlockingQueue} from '#common/async/blocking-queue.js'
 import {generateSignableJwt, jwkImport} from '#common/crypto/jwks'
 import {jwtPayload, verifyJwtToken} from '#common/crypto/jwts.js'
 import {normalizeError, normalizeProtocolError, ProtocolError} from '#common/errors'
···
   protocol.realmRtcPongResponseSchema,
 ])
 
-const realmRtcAutohandleSchema = z.union([
-  protocol.realmRtcPingRequestSchema,
-  protocol.realmRtcPongResponseSchema,
-])
-
 export interface ConnectionOptions {
   realmid?: RealmID
   register?: boolean
···
 
   #identity: RealmIdentity
   #sync: RealmSyncManager
-  #queue: BlockingQueue<{peer: RealmPeer; message: unknown}>
-  #queueAbort: AbortController
 
   constructor(
     url: string,
···
     this.#nonces = new Map()
 
     this.#sync = new RealmSyncManager(db, clock, identity.identid, this.#peers)
-
-    this.#queue = new BlockingQueue()
-    this.#queueAbort = new AbortController()
-    this.#queueRun(this.#queueAbort.signal).catch((err: unknown) => {
-      console.error('error in the queue runner!', err)
-    })
-
     this.#socket = new WebSocket(url)
     this.#socket.onopen = this.#handleSocketOpen
     this.#socket.onclose = this.#handleSocketClose
···
     const peer = this.#peers.get(identid)
     if (!peer) throw new Error(`Not connected to peer: ${identid}`)
 
-    console.debug('sending:', identid, data)
-    peer.send(JSON.stringify(data))
+    peer.sendJson(data)
   }
 
   broadcast(data: unknown, self = false) {
···
 
   destroy() {
     console.debug('realm connection destroy!')
-    this.#queueAbort.abort()
 
     if (this.connected) {
       this.#socket.close()
···
     this.dispatchEvent(new CustomEvent(type, {bubbles: true, detail}))
   }
 
-  async #queueRun(signal?: AbortSignal) {
-    signal?.throwIfAborted()
-    console.debug('realm message queue running!')
-
-    while (!signal?.aborted) {
-      const {peer, message} = await this.#queue.dequeue(signal)
-      const parsed = realmRtcAutohandleSchema.safeParse(message)
-      console.log('processing...', message, parsed)
-
-      // this is not autorespondable - pass along
-      if (!parsed.success) {
-        this.#dispatchCustomEvent('peerdata', {identid: peer.identid, data: message})
-        continue
-      }
-
-      // otherwise do the auto handling
-      switch (parsed.data.msg) {
-        case 'realm.rtc.ping':
-          await peer.incomingPing(parsed.data)
-          continue
-
-        case 'realm.rtc.pong':
-          await peer.incomingPong(parsed.data)
-          continue
-      }
-    }
-  }
-
   #socketSend<T extends object>(payload: T) {
     this.#socket.send(JSON.stringify(payload))
   }
···
       peer = new RealmPeer(this.#sync, remoteid, initiator)
       this.#peers.set(remoteid, peer)
 
+      // handle websocket signalling in here
       peer.on('signal', this.#handlePeerSignal.bind(this, peer))
-      peer.on('peermessage', this.#handlePeerMessage.bind(this))
+
+      // forward everything
+      peer.on('peerdata', this.dispatchEvent.bind(this))
       peer.on('peeropen', this.dispatchEvent.bind(this))
       peer.on('peerclose', this.dispatchEvent.bind(this))
       peer.on('peererror', this.dispatchEvent.bind(this))
···
       this.#peers.delete(identid)
       this.#nonces.delete(identid)
     }
-  }
-
-  #handlePeerMessage = (event: CustomEvent<{peer: RealmPeer; message: unknown}>) => {
-    this.#queue.enqueue(event.detail)
   }
 
   #handlePeerSignal = (peer: RealmPeer, payload: SimplePeer.SignalData) => {