# Patterns

> Three things every realtime client needs: matching command results to the commands you
> sent, ignoring repeated status updates, and catching up after a disconnect. And one
> thing Parklio doesn't have: webhooks.

# Realtime patterns

## Correlate results by `event_id`

[`POST /v2/products/{id}/state`](/api/endpoints/products-change-state) and
[`POST /v2/products/{id}/settings`](/api/endpoints/products-change-settings) return an
`event_id`. The result arrives later as a `lot_change` (success) or `gateway_error`
(failure) with the same `event_id`. See
[Asynchronous operations](/concepts/asynchronous-operations).

The helper below handles the cases that trip clients up:

- the result can arrive **before** the HTTP response, so unmatched results are kept for
  a while;
- your connection also gets results for operations you didn't send, which are ignored;
- a result may never come, so each wait has a timeout.

**Node.js**

```ts title="pending-operations.ts"
import type { Socket } from 'socket.io-client';

export interface OperationEvent {
  event_id?: string;
  type?: number | 'Timeout';
  message?: string;
  data: Record<string, any>;
}

export type OperationResult =
  | { ok: true; event: OperationEvent }
  | { ok: false; event: OperationEvent }
  | { ok: false; timedOut: true };

// Matches realtime results to the event_ids your commands returned.
export class PendingOperations {
  #waiting = new Map<string, { resolve: (r: OperationResult) => void; timer: NodeJS.Timeout }>();
  #early = new Map<string, OperationResult>(); // results that arrived before we asked

  constructor(socket: Socket, { keepEarlyMs = 60_000 } = {}) {
    const settle = (ok: boolean) => (raw: string) => {
      const event: OperationEvent = JSON.parse(raw);
      if (!event.event_id) return; // a status update, not a command result
      const eventId = event.event_id;
      const result: OperationResult = ok ? { ok: true, event } : { ok: false, event };
      const waiter = this.#waiting.get(eventId);
      if (waiter) {
        clearTimeout(waiter.timer);
        this.#waiting.delete(eventId);
        waiter.resolve(result);
      } else {
        this.#early.set(eventId, result);
        setTimeout(() => this.#early.delete(eventId), keepEarlyMs).unref();
      }
    };
    socket.on('lot_change', settle(true));
    socket.on('gateway_error', settle(false));
  }

  // Resolves with { ok, event }, or { ok: false, timedOut: true }.
  wait(eventId: string, timeoutMs: number): Promise<OperationResult> {
    const early = this.#early.get(eventId);
    if (early) {
      this.#early.delete(eventId);
      return Promise.resolve(early);
    }
    return new Promise((resolve) => {
      const timer = setTimeout(() => {
        this.#waiting.delete(eventId);
        resolve({ ok: false, timedOut: true });
      }, timeoutMs);
      this.#waiting.set(eventId, { resolve, timer });
    });
  }
}
```

```ts title="Usage"
const pending = new PendingOperations(socket); // create it right after connecting

const res = await fetch(`https://api.parklio.com/v2/products/${productId}/state`, {
  method: 'POST',
  headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' },
  body: JSON.stringify({ operation: 'open' }),
});
const { data } = await res.json();
const result = await pending.wait(data.event_id, 120_000);
```

**Python**

```python title="pending_operations.py"
import json
import threading
import time

class PendingOperations:
    """Matches realtime results to the event_ids your commands returned."""

    def __init__(self, sio, namespace="/v2/realtime", keep_early_s=60):
        self._lock = threading.Lock()
        self._waiting = {}  # event_id -> [threading.Event, result]
        self._early = {}  # event_id -> (result, arrived_at)
        self._keep_early_s = keep_early_s
        sio.on("lot_change", self._settle(True), namespace=namespace)
        sio.on("gateway_error", self._settle(False), namespace=namespace)

    def _settle(self, ok):
        def handler(raw):
            event = json.loads(raw)
            event_id = event.get("event_id")
            if not event_id:
                return  # a status update, not a command result
            result = {"ok": ok, "event": event}
            with self._lock:
                slot = self._waiting.get(event_id)
                if slot:
                    slot[1] = result
                    slot[0].set()
                else:
                    now = time.monotonic()
                    self._early = {
                        k: v for k, v in self._early.items() if now - v[1] < self._keep_early_s
                    }
                    self._early[event_id] = (result, now)
        return handler

    def wait(self, event_id, timeout_s):
        """Returns {"ok", "event"}, or {"ok": False, "timed_out": True}."""
        with self._lock:
            early = self._early.pop(event_id, None)
            if early:
                return early[0]
            slot = self._waiting[event_id] = [threading.Event(), None]
        slot[0].wait(timeout_s)
        with self._lock:
            self._waiting.pop(event_id, None)
        return slot[1] or {"ok": False, "timed_out": True}
```

```python title="Usage"
pending = PendingOperations(sio)  # create it before connecting

res = httpx.post(
    f"https://api.parklio.com/v2/products/{product_id}/state",
    headers={"Authorization": f"Bearer {token}"},
    json={"operation": "open"},
)
result = pending.wait(res.json()["data"]["event_id"], timeout_s=120)
```

When a wait times out, the operation may still have run. Re-read the product with
[`GET /v2/products/{id}`](/api/endpoints/products-find-one) before deciding what to do.

## Deduplicate status updates

`lot_change` events **without** an `event_id` come from Gateways reporting what they see
over Bluetooth. Two things to know:

- **Each one is sent twice**, with identical payloads.
- **They repeat even when nothing changed**: a Gateway's report re-sends every product it
  sees.

So treat them as "current state of product X", not as "product X changed". Keep the last
state per product, and act only when something you care about differs:

```ts title="Ignore repeats"
const lastState = new Map<number, string>(); // product id -> fingerprint

socket.on('lot_change', (raw: string) => {
  const { event_id, data } = JSON.parse(raw);
  if (event_id) return; // command results are handled by PendingOperations
  const block = data.barrier ?? data.gate ?? data.chain ?? data.bollard ?? data.brain;
  const fingerprint = JSON.stringify([data.status, block?.state, data.product_error]);
  if (lastState.get(data.id) === fingerprint) return;
  lastState.set(data.id, fingerprint);
  onProductChanged(data); // your code
});
```

Pick the fields for the fingerprint from what you act on; `battery_status` and `rssi`
change often and are rarely worth an alert.

## Resync after a reconnect

There's no replay: anything sent while you were disconnected is gone. So every time you
connect, re-read the state you track over REST, page by page. That covers the first
connect too, where events sent in the first moment can be missed (see
[Authentication](/realtime/connecting#authentication)).

```ts title="Resync products of a lot on every connect"
socket.on('connect', () => resyncLot(lotId)); // fires on the first connect and after each reconnect

async function resyncLot(lotId: number) {
  for (let page = 1, last = 1; page <= last; page++) {
    const url = `https://api.parklio.com/v2/products?filter=lot_id:eq:${lotId}&page=${page}&size=100`;
    const body = await (await fetch(url, { headers: { Authorization: `Bearer ${token}` } })).json();
    for (const product of body.data) {
      lastState.delete(product.id);
      onProductChanged(product); // note: REST and realtime product shapes differ slightly
    }
    last = body.meta.last_page;
  }
}
```

Also treat every command that was pending during the disconnect as unknown: re-read
those products too.

In Python, `python-socketio` calls your `connect` handler on the first connect and after
each reconnect too; start the resync from there.

## No webhooks

Parklio doesn't call your URLs. Realtime events are delivered only over this Socket.IO
connection, to clients that are connected at the time. If other systems need the
events, run one small service that holds the connection and forwards what you need (to
a queue, a database or your own webhooks), and resyncs after reconnects as above.

## Next

- [Events](/realtime/events): every event and its payload.
- [Asynchronous operations](/concepts/asynchronous-operations)