# @causal-order/transport

WebSocket + JSON ingress with acknowledged delivery, bounded shutdown, and
event normalization for the [`causal-order`](https://www.npmjs.com/package/causal-order) stack.

Published package version: `v0.2.2`

## Stack Position

```text
nodes -> @causal-order/transport -> @causal-order/dedupe -> causal-order -> application
```

Transport owns WebSocket connection lifecycle, the versioned JSON protocol,
delivery acknowledgment, peer evidence, and normalization into the canonical
event contract. Deduplication and causal ordering remain downstream concerns.

Deployments requiring bounded outage buffering, health-aware routing,
controlled recovery, and replay place
[`@causal-order/monitor`](https://www.npmjs.com/package/@causal-order/monitor)
between transport and dedupe:

```text
nodes -> @causal-order/transport -> @causal-order/monitor -> @causal-order/dedupe -> causal-order -> application
```

Monitor owns buffering and replay. Replayed events return through dedupe before
causal ordering.

## Install and Requirements

```bash
npm install @causal-order/transport @causal-order/dedupe causal-order
```

- Supported Node.js release lines: 22 and 24
- ESM-only output

The monitor-enabled stack requires Node.js 22.13.0 or newer because
`@causal-order/monitor` uses the built-in `node:sqlite` module.

## Server

Register one primary delivery authority when downstream acceptance must control
the acknowledgment returned to the sender.

```ts
import { WebSocketJsonTransport } from "@causal-order/transport";

const transport = new WebSocketJsonTransport({
  mode: "server",
  host: "127.0.0.1",
  port: 8080,
});

transport.onDelivery(async (event, context) => {
  await deliverToNextStage(event, context);
  return { outcome: "accepted" };
});

transport.onEvent((event, context) => {
  observeTransportEvent(event, context);
});

transport.onPeerState((state) => {
  console.log("peer", state.peerId, state.status);
});

transport.onError((error) => {
  console.error("transport", error.detail);
});

await transport.start();
```

`onDelivery()` is the delivery authority. `onEvent()` is optional observation
and does not control acknowledgment. `deliverToNextStage` is
application-provided; it normally hands the event to monitor or directly to
dedupe. Without an `onDelivery()` handler, successful normalization and
observer dispatch form the acknowledgment boundary.

The default normalizer is used when `normalizeMessage` is omitted. A custom
normalizer may adapt another inbound JSON event shape:

```ts
const transport = new WebSocketJsonTransport({
  mode: "server",
  port: 8080,
  normalizeMessage(message) {
    return normalizeApplicationMessage(message);
  },
});
```

## Client

Client transports accept normalized events containing bigint-backed causal
metadata. Known bigint fields are encoded as decimal strings on the JSON wire.
`send()` resolves only after the receiving transport's primary delivery handler
acknowledges the transmission. A WebSocket write callback alone is not treated
as remote delivery.

```ts
import { WebSocketJsonTransport } from "@causal-order/transport";

const client = new WebSocketJsonTransport({
  mode: "client",
  url: "ws://127.0.0.1:8080",
  peerId: "ingress",
});

await client.start();

await client.send({
  id: "edge-a-000000000042",
  nodeId: "edge-a",
  sequence: 42n,
  clock: {
    physicalTimeMs: 1781000000000n,
    logicalCounter: 0,
    nodeId: "edge-a",
  },
  parentEventId: "edge-a-000000000041",
  dependencyEventIds: ["edge-b-000000000017"],
  partition: "factory-1",
  payload: { temperature: 21 },
});
```

Payloads must be JSON-serializable. Arbitrary bigint values inside `payload`
are not converted automatically.

In server mode, `send(event)` broadcasts to all connected peers.
`send(event, peerId)` targets one connected peer.

Every unicast target receives a unique transmission ID. Broadcasts snapshot
their target peers and wait for every target to acknowledge or fail. A partial
broadcast failure rejects with a `TransportOperationError` whose cause retains
the individual peer failures.

## Default Wire Format

Transport `0.2.x` sends a versioned event envelope:

```json
{
  "schema": "causal-order/transport",
  "version": 1,
  "type": "event",
  "transmissionId": "tx-018f-example",
  "event": {
    "id": "edge-a-000000000042",
    "nodeId": "edge-a",
    "sequence": "42",
    "clock": {
      "physicalTimeMs": "1781000000000",
      "logicalCounter": 0,
      "nodeId": "edge-a"
    },
    "parentEventId": "edge-a-000000000041",
    "dependencyEventIds": ["edge-b-000000000017"],
    "partition": "factory-1",
    "payload": {
      "temperature": 21
    }
  }
}
```

The receiver returns a correlated result after primary delivery:

```json
{
  "schema": "causal-order/transport",
  "version": 1,
  "type": "delivery-result",
  "transmissionId": "tx-018f-example",
  "outcome": "accepted"
}
```

Refusal and failure results carry the same `transmissionId`, a non-accepted
outcome, and a machine-readable code. Application event IDs do not serve as
transmission IDs, so retries and intentional duplicates remain independently
accountable.

The receiver continues to accept the legacy flat `0.1.x` event object as input,
but legacy messages cannot participate in acknowledged root `send()` semantics.

The following aliases are also accepted:

| Canonical field | Accepted aliases |
| --- | --- |
| `nodeId` | `node`, `sourceNodeId` |
| `sequence` | `seq` |
| `clock.physicalTimeMs` | `ts`, `timestampMs`, `physicalTimeMs` |
| `clock.logicalCounter` | `logical`, `lc` |
| `payload` | `body` |

`sequence`, timestamps, and `ingestedAt` accept bigint values, safe integer
numbers, or canonical decimal integer strings. Unsafe numbers and malformed
integer strings are rejected.

Normalized events contain:

- `id`
- `nodeId`
- optional `sequence`
- `clock.physicalTimeMs`, `clock.logicalCounter`, and `clock.nodeId`
- `payload`
- optional `partition`, `parentEventId`, and `dependencyEventIds`
- optional `traceId` and `ingestedAt`

## API

The main entry point, `@causal-order/transport`, exports:

- `WebSocketJsonTransport`
- `normalizeTransportEventMessage()`
- `createEventId()`
- `createHarnessAdapter()`
- transport, event, context, peer-state, and option types

The published testing adapter supports deterministic node disconnect/reconnect
and accepts the root transport's bounded per-peer send capacity:

```ts
const adapter = await createHarnessAdapter({
  nodeIds: ["edge-a", "edge-b"],
  maxInFlightSendsPerPeer: 4096,
});

await adapter.setNodeConnectivity("edge-a", "disconnected");
const bufferedSend = adapter.send(event);
await adapter.setNodeConnectivity("edge-a", "connected");
await bufferedSend;
```

`WebSocketJsonTransport` implements this lifecycle:

```ts
interface TransportContract<T> {
  readonly state: TransportLifecycleState;
  start(): Promise<void>;
  stop(): Promise<void>;
  send(event: NormalizedTransportEvent<T>, targetPeerId?: string): Promise<void>;
  onDelivery(handler: TransportDeliveryHandler<T>): () => void;
  onEvent(handler: TransportEventHandler<T>): () => void;
  onPeerState(handler: TransportPeerStateHandler): () => void;
  onError(handler: TransportErrorHandler): () => void;
}
```

`onDelivery()` registers one primary delivery authority. Its result determines
the correlated acknowledgment returned to the sender. It may be asynchronous
and may accept or refuse delivery. `onEvent()` remains an observational
notification; observer failures are reported without changing the primary
delivery result.

Lifecycle state is `idle`, `starting`, `started`, `stopping`, `stopped`, or
`failed`. Shutdown atomically rejects new sends, drains accepted sends and
receive work, closes sockets, and resolves only when no transport callback can
arrive later. A bounded drain failure rejects explicitly.

Timeout and pressure options include `connectionTimeoutMs`, `acknowledgmentTimeoutMs`,
`deliveryTimeoutMs`, `shutdownTimeoutMs`, `maxInFlightSends`,
`maxInFlightSendsPerPeer`, `maxInFlightReceives`,
`maxInFlightReceivesPerPeer`, and `maxBufferedAmountBytes`.

Instances may restart after a completed stop. Automatic reconnect is not part
of `0.2.1`; applications must create or restart a client according to their own
retry policy.

## Testing Subpath

The package exports `@causal-order/transport/testing` for
[`@causal-order/testing`](https://www.npmjs.com/package/@causal-order/testing).
It is validation tooling, not an application runtime API.

```bash
npm install --save-dev @causal-order/testing
```

The adapter uses root transport clients and exercises the same acknowledgment,
pressure, and stop-barrier contract as the production API. Repository test
profiles and release evidence are documented in
[`VALIDATION.md`](https://github.com/GazaliAhmad/causal-order-transport/blob/main/VALIDATION.md).

## Runtime Boundary

- This package provides transport and normalization. Deduplication and causal
  ordering remain the responsibility of `@causal-order/dedupe` and
  `causal-order`.
- WebSocket + JSON is the supported ingress path. Other wire protocols should
  use separate adapters built against the same normalized event contract.
- The required `ws` runtime is vendored behind an internal wrapper, so consumers
  do not install `ws` as an additional runtime dependency.
- Vendored source, license, version, and local-modification details are recorded
  in [`THIRD_PARTY_NOTICES.md`](https://github.com/GazaliAhmad/causal-order-transport/blob/main/THIRD_PARTY_NOTICES.md).

## License

[MIT](https://github.com/GazaliAhmad/causal-order-transport/blob/main/LICENSE)