[DO NOT MERGE] refactor(sdk): network layer foundations — break cycles, single isServer, logging discipline

packages/@dcl/sdk/src/internal/transports/logger.ts

@@ -10,9 +10,6 @@ export function* serializeCrdtMessages(prefix: string, data: Uint8Array, engine:
   while ((message = readMessage(buffer))) {
     const ent = message.entityId
     const preface = `${prefix}: ${CrdtMessageType[message.type]} e=${ent}`
-    if (message.type === CrdtMessageType.DELETE_ENTITY || message.type === CrdtMessageType.DELETE_ENTITY_NETWORK) {
-      yield `${preface}`
-    }
 
     if (
       message.type === CrdtMessageType.PUT_COMPONENT ||

packages/@dcl/sdk/src/internal/transports/rendererTransport.ts

@@ -28,7 +28,6 @@ export function createRendererTransport(engineApi: EngineApiForTransport): Trans
         // this is the console.error of the scene
         // eslint-disable-next-line no-console
         console.error(error)
-        debugger
       }
     },
     filter(message: TransportMessage) {

packages/@dcl/sdk/src/network/binary-message-bus.ts

@@ -1,4 +1,4 @@
-import { ReadWriteByteBuffer } from '@dcl/ecs/dist/serialization/ByteBuffer'
+import { ReadWriteByteBuffer } from './ecs-adapter'
 
 export enum CommsMessage {
   CRDT = 7,

packages/@dcl/sdk/src/network/chunking.ts

@@ -1,4 +1,4 @@
-import { ReadWriteByteBuffer } from '@dcl/ecs/dist/serialization/ByteBuffer'
+import { ReadWriteByteBuffer } from './ecs-adapter'
 import { readMessages } from './server/utils'
 
 /**

packages/@dcl/sdk/src/network/constants.ts

@@ -0,0 +1,12 @@
+// Leaf module: it must not import anything from `network/`. Holding these here is
+// what keeps `message-bus-sync → state → server` acyclic.
+
+export type IProfile = { networkId: number; userId: string }
+
+/** peer we ask the initial CRDT state from */
+export const AUTH_SERVER_PEER_ID = 'authoritative-server'
+
+export const DEBUG_NETWORK_MESSAGES = () => (globalThis as any).DEBUG_NETWORK_MESSAGES ?? false
+
+/** max payload livekit accepts, in KB */
+export const LIVEKIT_MAX_SIZE = 12

packages/@dcl/sdk/src/network/ecs-adapter.ts

@@ -0,0 +1,42 @@
+// Single place where the network layer reaches into `@dcl/ecs/dist/**` internals.
+// Promoting these to the public `@dcl/ecs` API has to clear that package's 100%
+// coverage gate, so it is a separate PR; until then every deep path lives here
+// instead of being spread over the layer.
+
+export * as components from '@dcl/ecs/dist/components'
+export { ReadWriteByteBuffer } from '@dcl/ecs/dist/serialization/ByteBuffer'
+export type { ByteBuffer } from '@dcl/ecs/dist/serialization/ByteBuffer'
+export { componentNumberFromName } from '@dcl/ecs/dist/components/component-number'
+
+export { AuthoritativePutComponentOperation, PutComponentOperation } from '@dcl/ecs/dist/serialization/crdt'
+export { DeleteComponent } from '@dcl/ecs/dist/serialization/crdt/deleteComponent'
+export { DeleteEntity } from '@dcl/ecs/dist/serialization/crdt/deleteEntity'
+export { PutNetworkComponentOperation } from '@dcl/ecs/dist/serialization/crdt/network/putComponentNetwork'
+export { DeleteComponentNetwork } from '@dcl/ecs/dist/serialization/crdt/network/deleteComponentNetwork'
+export { DeleteEntityNetwork } from '@dcl/ecs/dist/serialization/crdt/network/deleteEntityNetwork'
+export { TransformSchema, COMPONENT_ID as TransformComponentId } from '@dcl/ecs/dist/components/manual/Transform'
+
+export type {
+  CrdtMessage,
+  CrdtMessageBody,
+  CrdtMessageHeader,
+  DeleteComponentMessage,
+  DeleteComponentNetworkMessage,
+  DeleteEntityMessage,
+  DeleteEntityNetworkMessage,
+  PutComponentMessage,
+  AuthoritativePutComponentMessage,
+  PutNetworkComponentMessage
+} from '@dcl/ecs/dist/serialization/crdt/types'
+export type { ReceiveMessage } from '@dcl/ecs/dist/runtime/types'
+export type { ReceiveNetworkMessage } from '@dcl/ecs/dist/systems/crdt/types'
+export type { INetowrkEntityType } from '@dcl/ecs/dist/components/types'
+
+// `__dry_run_updateFromCrdt` / `__run_validateBeforeChange` only exist on these
+// internal shapes, which is why the validator needs them.
+export type {
+  LastWriteWinElementSetComponentDefinition,
+  GrowOnlyValueSetComponentDefinition,
+  ComponentDefinition,
+  InternalBaseComponent
+} from '@dcl/ecs/dist/engine/component'

packages/@dcl/sdk/src/network/entities.ts

@@ -10,7 +10,7 @@ import {
   TransformComponent,
   ISyncComponents
 } from '@dcl/ecs'
-import { IProfile } from './message-bus-sync'
+import { IProfile } from './constants'
 import { getDesyncedComponents } from './state'
 
 export type SyncEntity = (entityId: Entity, componentIds: number[], entityEnumId?: number) => void

packages/@dcl/sdk/src/network/events/implementation.ts

@@ -1,10 +1,9 @@
 import { IEngine } from '@dcl/ecs'
 import { CommsMessage } from '../binary-message-bus'
-import { AUTH_SERVER_PEER_ID } from '../message-bus-sync'
+import { AUTH_SERVER_PEER_ID } from '../constants'
 import { EventTypes, EventSchemaRegistry } from './registry'
 import { encodeEvent, decodeEvent } from './protocol'
 import { Atom } from '../../atom'
-import { future, IFuture } from '../../future'
 
 // Context provided to server-side event handlers
 export type EventContext = {
@@ -34,14 +33,13 @@ type QueuedMessage<T extends EventSchemaRegistry = EventSchemaRegistry> = {
 export class Room<T extends EventSchemaRegistry = EventSchemaRegistry> {
   private listeners = new Map<keyof T, Set<EventCallback<any>>>()
   private binaryMessageBus: any
-  private isServerFuture: IFuture<boolean> = future()
+  private isServerAtom: Atom<boolean>
   private isRoomReadyAtom: Atom<boolean>
   private messageQueue: QueuedMessage<T>[] = []
   private isProcessingQueue = false
 
-  constructor(_engine: IEngine, binaryMessageBus: any, isServerFn: Atom<boolean>, isRoomReadyAtom: Atom<boolean>) {
-    void isServerFn.deref().then(($) => this.isServerFuture.resolve($))
-
+  constructor(_engine: IEngine, binaryMessageBus: any, isServerAtom: Atom<boolean>, isRoomReadyAtom: Atom<boolean>) {
+    this.isServerAtom = isServerAtom
     this.binaryMessageBus = binaryMessageBus
     this.isRoomReadyAtom = isRoomReadyAtom
 
@@ -59,7 +57,7 @@ export class Room<T extends EventSchemaRegistry = EventSchemaRegistry> {
 
         if (callbacks) {
           callbacks.forEach(async (cb) => {
-            if (await this.isServerFuture) {
+            if (await this.isServerAtom.deref()) {
               // Server handlers receive sender context
               cb(payload, { from: sender })
             } else if (sender === AUTH_SERVER_PEER_ID) {
@@ -119,7 +117,7 @@ export class Room<T extends EventSchemaRegistry = EventSchemaRegistry> {
       // Room is ready, send immediately
       const buffer = encodeEvent(eventType as string, data, globalEventRegistry)
 
-      if (await this.isServerFuture) {
+      if (await this.isServerAtom.deref()) {
         // Server can send to specific clients or broadcast
         this.binaryMessageBus.emit(CommsMessage.CUSTOM_EVENT, buffer, options?.to)
       } else {
@@ -247,8 +245,6 @@ export function registerMessages<T extends EventSchemaRegistry>(messages: T): Ro
   if (!globalRoom) {
     throw new Error('Room not initialized. Make sure the SDK network transport is initialized.')
   }
-  // Update the room registry
-  ;(globalRoom as any).registry = globalEventRegistry
   return globalRoom as unknown as Room<T>
 }
 

packages/@dcl/sdk/src/network/events/protocol.ts

@@ -1,4 +1,4 @@
-import { ReadWriteByteBuffer } from '@dcl/ecs/dist/serialization/ByteBuffer'
+import { ReadWriteByteBuffer } from '../ecs-adapter'
 import { Schemas } from '@dcl/ecs'
 import { EventSchemas, EventTypes, EventSchemaRegistry } from './registry'
 

packages/@dcl/sdk/src/network/index.ts

@@ -3,18 +3,6 @@ import { engine } from '@dcl/ecs'
 import { addSyncTransport } from './message-bus-sync'
 import { getUserData } from '~system/UserIdentity'
 import { isServer as isServerApi } from '~system/EngineApi'
-import { Atom } from '../atom'
-
-// Create isServer atom for consistent state
-const isServerAtom = Atom<boolean>(false)
-void isServerApi({}).then((response) => {
-  isServerAtom.swap(!!response.isServer)
-})
-
-// Helper function to check if running on server
-export function isServer(): boolean {
-  return isServerAtom.getOrNull() ?? false
-}
 
 // initialize sync transport for sdk engine
 const {
@@ -27,9 +15,15 @@ const {
   getFirstChild,
   isStateSyncronized,
   binaryMessageBus,
-  eventBus
+  eventBus,
+  isServerAtom
 } = addSyncTransport(engine, sendBinary, getUserData, isServerApi, 'network')
 
+// Helper function to check if running on server
+export function isServer(): boolean {
+  return isServerAtom.getOrNull() ?? false
+}
+
 // Re-export the room messaging system
 export { registerMessages, getRoom } from './events'
 

packages/@dcl/sdk/src/network/message-bus-sync.ts

@@ -11,13 +11,12 @@ import { GetUserDataRequest, GetUserDataResponse } from '~system/UserIdentity'
 import { definePlayerHelper } from '../players'
 import { serializeCrdtMessages } from '../internal/transports/logger'
 import { IsServerRequest, IsServerResponse } from '~system/EngineApi'
-import { Atom } from '../atom'
+import { AUTH_SERVER_PEER_ID, DEBUG_NETWORK_MESSAGES, IProfile } from './constants'
+import { createRuntimeContext } from './runtime-context'
 import { setGlobalRoom, Room } from './events/implementation'
 
-export type IProfile = { networkId: number; userId: string }
-// user that we asked for the inital crdt state
-export const AUTH_SERVER_PEER_ID = 'authoritative-server'
-export const DEBUG_NETWORK_MESSAGES = () => (globalThis as any).DEBUG_NETWORK_MESSAGES ?? false
+export { AUTH_SERVER_PEER_ID, DEBUG_NETWORK_MESSAGES } from './constants'
+export type { IProfile } from './constants'
 
 // Test environment detection without 'as any'
 const isTestEnvironment = (): boolean => {
@@ -41,12 +40,7 @@ export function addSyncTransport(
   const myProfile: IProfile = {} as IProfile
   fetchProfile(myProfile!, getUserData)
 
-  const isServerAtom = Atom<boolean>()
-  const isRoomReadyAtom = Atom<boolean>(false)
-
-  void isServerFn({}).then(($: IsServerResponse) => {
-    return isServerAtom.swap(!!$.isServer)
-  })
+  const { isServerAtom, isRoomReadyAtom } = createRuntimeContext(isServerFn)
 
   // Entity utils
   const entityDefinitions = entityUtils(engine, myProfile)
@@ -98,6 +92,16 @@ export function addSyncTransport(
     type: name
   }
 
+  // `onmessage` is wired by `engine.addTransport`; a comms message that arrives
+  // before that (or after a transport swap) must not take the handler down.
+  function deliverToEngine(buffer: Uint8Array) {
+    if (!transport.onmessage) {
+      DEBUG_NETWORK_MESSAGES() && console.log('[deliverToEngine] transport not wired yet, dropping', buffer.byteLength)
+      return
+    }
+    transport.onmessage(buffer)
+  }
+
   // Server validation setup
   const serverValidator = createServerValidator({
     engine,
@@ -133,7 +137,7 @@ export function addSyncTransport(
     if (isServerAtom.getOrNull() || sender !== AUTH_SERVER_PEER_ID) return
     DEBUG_NETWORK_MESSAGES() && console.log('[Processing CRDT State]', data.byteLength / 1024, 'KB')
     if (data.byteLength > 0) {
-      transport.onmessage!(serverValidator.processClientMessages(data, sender))
+      deliverToEngine(serverValidator.processClientMessages(data, sender))
     }
     stateIsSyncronized = true
 
@@ -156,10 +160,10 @@ export function addSyncTransport(
         isServer
       )
     if (isServer) {
-      transport.onmessage!(serverValidator.processServerMessages(value, sender))
+      deliverToEngine(serverValidator.processServerMessages(value, sender))
     } else if (sender === AUTH_SERVER_PEER_ID) {
       // Process network messages from server and convert to regular messages
-      transport.onmessage!(serverValidator.processClientMessages(value, sender))
+      deliverToEngine(serverValidator.processClientMessages(value, sender))
     }
   })
 
@@ -168,16 +172,16 @@ export function addSyncTransport(
     // Only accept authoritative messages from authoritative server
     if (sender !== AUTH_SERVER_PEER_ID) return
 
-    // DEBUG_NETWORK_MESSAGES() &&
-    console.log('[AUTHORITATIVE] Received authoritative message from server:', value.byteLength, 'bytes')
+    DEBUG_NETWORK_MESSAGES() &&
+      console.log('[AUTHORITATIVE] Received authoritative message from server:', value.byteLength, 'bytes')
 
     // Process authoritative messages by forcing them through normal CRDT processing
     // but with a timestamp that's guaranteed to be accepted
     const authoritativeBuffer = serverValidator.processClientMessages(value, sender, true)
     if (authoritativeBuffer.byteLength > 0) {
       // Apply authoritative message through normal transport, but the server's messages
       // should be processed as authoritative with special timestamp handling
-      transport.onmessage!(authoritativeBuffer)
+      deliverToEngine(authoritativeBuffer)
 
       DEBUG_NETWORK_MESSAGES() && console.log('[AUTHORITATIVE] Applied server authoritative message to local state')
     }
@@ -267,6 +271,7 @@ export function addSyncTransport(
     isStateSyncronized,
     binaryMessageBus,
     eventBus,
+    isServerAtom,
     isRoomReadyAtom
   }
 }

packages/@dcl/sdk/src/network/runtime-context.ts

@@ -0,0 +1,23 @@
+import { IsServerRequest, IsServerResponse } from '~system/EngineApi'
+import { Atom } from '../atom'
+
+/**
+ * Per-transport runtime state. Built once per `addSyncTransport` rather than at
+ * module scope on purpose: several transports (each with its own `isServerFn`)
+ * coexist in one process in the tests, and they must not share a role.
+ */
+export type RuntimeContext = {
+  isServerAtom: Atom<boolean>
+  isRoomReadyAtom: Atom<boolean>
+}
+
+export function createRuntimeContext(
+  isServerFn: (request: IsServerRequest) => Promise<IsServerResponse>
+): RuntimeContext {
+  const isServerAtom = Atom<boolean>()
+  const isRoomReadyAtom = Atom<boolean>(false)
+
+  void isServerFn({}).then(($: IsServerResponse) => isServerAtom.swap(!!$.isServer))
+
+  return { isServerAtom, isRoomReadyAtom }
+}

packages/@dcl/sdk/src/network/server/index.ts

@@ -7,21 +7,21 @@ import {
   ComponentType,
   PutNetworkComponentOperation
 } from '@dcl/ecs'
-import * as components from '@dcl/ecs/dist/components'
-import { ReadWriteByteBuffer } from '@dcl/ecs/dist/serialization/ByteBuffer'
 import { CommsMessage } from '../binary-message-bus'
 import { chunkCrdtMessages } from '../chunking'
 import * as utils from './utils'
-import { AUTH_SERVER_PEER_ID, DEBUG_NETWORK_MESSAGES } from '../message-bus-sync'
+import { AUTH_SERVER_PEER_ID, DEBUG_NETWORK_MESSAGES, LIVEKIT_MAX_SIZE } from '../constants'
 import { type BinaryMessageBus } from '../binary-message-bus'
 import {
+  components,
+  ReadWriteByteBuffer,
   LastWriteWinElementSetComponentDefinition,
   GrowOnlyValueSetComponentDefinition,
   ComponentDefinition,
   InternalBaseComponent
-} from '@dcl/ecs/dist/engine/component'
+} from '../ecs-adapter'
 
-export const LIVEKIT_MAX_SIZE = 12
+export { LIVEKIT_MAX_SIZE } from '../constants'
 
 export interface ServerValidationConfig {
   engine: IEngine
@@ -100,7 +100,7 @@ export function createServerValidator(config: ServerValidationConfig) {
       )
       return { ...message, messageBuffer: buffer.toBinary() }
     } catch (error) {
-      DEBUG_NETWORK_MESSAGES() && console.error('Error converting network message:', error)
+      console.error('Error converting network message:', error)
       return null
     }
   }
@@ -272,7 +272,7 @@ export function createServerValidator(config: ServerValidationConfig) {
             }
           }
         } catch (error) {
-          DEBUG_NETWORK_MESSAGES() && console.error('Error processing server message:', error)
+          console.error('Error processing server message:', error)
         }
       }
       // Batch broadcast all valid messages together