Method pull

src/proxy/sync/inbound.js:61–142

(channel = 'evomap-hub', limit = 50)

  }

  async pull(channel = 'evomap-hub', limit = 50) {
    const waitMs = this._hubUnreachableWaitMs();
    if (waitMs > 0) {
      return {
        received: 0,
        error: 'hub_unreachable_backoff',
        hubUnreachable: true,
        retryAfterMs: waitMs,
      };
    }

    const cursorKey = `${channel}:inbound_cursor`;
    const cursor = this.store.getCursor(cursorKey);

    const endpoint = `${this.hubUrl}/a2a/mailbox/inbound`;

    try {
      const senderId = this.store.getState('node_id');
      const res = await hubFetch(endpoint, {
        method: 'POST',
        headers: this.getHeaders(),
        body: JSON.stringify({ sender_id: senderId, proxy_protocol_version: PROXY_PROTOCOL_VERSION, cursor, limit }),
        signal: AbortSignal.timeout(35_000),
      });

      await throwIfHubUnreachableResponse(res, 'inbound pull');
      this._recordHubReachable();

      if (res.status === 403 || res.status === 401) {
        const errText = await readHubResponseText(res).catch(() => 'unknown');
        throw new AuthError(`Hub ${res.status}: ${sanitizeHubResponseForLog(errText)}`, res.status);
      }

      if (!res.ok) {
        const errText = await readHubResponseText(res).catch(() => 'unknown');
        throw new Error(`Hub returned ${res.status}: ${sanitizeHubResponseForLog(errText)}`);
      }

      const data = await readHubResponseJson(res);
      const messages = data.messages || [];
      let stored = 0;
      let dropped = 0;

      if (messages.length > 0) {
        for (const m of messages) {
          try {
            this.store.writeInbound(toInboundStoreMessage(m, channel));
            stored++;
          } catch (err) {
            if (err && err.code === 'MAILBOX_JSONL_LINE_TOO_LARGE') {
              dropped++;
              const msgId = m && m.id ? String(m.id) : '(missing id)';
              this.logger.warn?.(`[inbound] dropped oversized inbound message ${msgId}: ${err.message}`);
              continue;
            }
            throw err;
          }
        }