feat: implement polling fallback for event communication

[takaokouji]

Summary

Client-side implementation of polling-based event notification fallback for Mesh v2. Adds WebSocket connection testing and automatic protocol switching to support environments where WebSocket connections are blocked.

Implementation Details

WebSocket Connection Test

Added testWebSocket() method to detect WebSocket availability:

testWebSocket() {
  return new Promise(resolve => {
    try {
      const wsUrl = GRAPHQL_ENDPOINT
        .replace('https://', 'wss://')
        .replace('appsync-api', 'appsync-realtime-api');

      const socket = new WebSocket(wsUrl, 'graphql-ws');
      const timeout = setTimeout(() => {
        socket.close();
        resolve(false);
      }, 3000);

      socket.onopen = () => {
        clearTimeout(timeout);
        socket.close();
        resolve(true);
      };

      socket.onerror = () => {
        clearTimeout(timeout);
        resolve(false);
      };
    } catch (error) {
      resolve(false);
    }
  });
}

Protocol Selection

Host (Group Creation)

1. Tests WebSocket connectivity using testWebSocket()
2. Sets useWebSocket flag based on test result
3. Sends flag to server via createGroup mutation
4. Starts subscription or polling based on flag

Member (Group Join)

1. Receives useWebSocket flag from joinGroup response
2. Starts subscription or polling based on flag

Polling Implementation

Start Polling

startPolling() {
  this.stopPolling();
  if (!this.groupId) return;

  log.info(`Mesh V2: Starting event polling (Interval: ${this.pollingIntervalSeconds}s)`);

  if (!this.lastFetchTime) {
    this.lastFetchTime = new Date().toISOString();
  }

  this.pollingTimer = setInterval(() => {
    this.pollEvents();
  }, this.pollingIntervalSeconds * 1000);
}

Poll Events

async pollEvents() {
  if (!this.groupId || !this.client || this.useWebSocket) return;

  try {
    const result = await this.client.query({
      query: GET_EVENTS_SINCE,
      variables: {
        groupId: this.groupId,
        domain: this.domain,
        since: this.lastFetchTime
      },
      fetchPolicy: 'network-only'
    });

    const events = result.data.getEventsSince;
    if (events && events.length > 0) {
      log.info(`Mesh V2: Polled ${events.length} events`);

      const batchEvent = {
        firedByNodeId: 'polling-server',
        events: events.map(e => ({ /* ... */ }))
      };
      this.handleBatchEvent(batchEvent);

      // Update cursor
      this.lastFetchTime = events[events.length - 1].cursor;
    }
  } catch (error) {
    log.error(`Mesh V2: Event polling failed: ${error}`);
    const reason = this.shouldDisconnectOnError(error);
    if (reason) {
      this.cleanupAndDisconnect(reason);
    }
  }
}

Event Transmission Protocol Switching

if (this.useWebSocket) {
  // Protocol A: WebSocket
  await this.client.mutate({
    mutation: FIRE_EVENTS,
    variables: { /* ... */ }
  });
} else {
  // Protocol B: Polling
  const result = await this.client.mutate({
    mutation: RECORD_EVENTS,
    variables: { /* ... */ }
  });

  // Update lastFetchTime if null
  if (!this.lastFetchTime) {
    this.lastFetchTime = result.data.recordEventsByNode.nextSince;
  }
}

Cursor Management

• lastFetchTime: Stores last cursor (or initial ISO timestamp)
• Updated to Event.cursor after each polling
• Used as since parameter in next getEventsSince call

GraphQL Operations

Added new operations to gql-operations.js:

• RECORD_EVENTS: Mutation for recording events (polling)
• GET_EVENTS_SINCE: Query for retrieving events (polling)

Testing

Unit Tests

New test file: test/unit/mesh_service_v2_polling.js

• pollEvents fetches and handles events correctly
• fireEventsBatch uses RECORD_EVENTS when useWebSocket is false
• Cursor-based pagination works correctly
• lastFetchTime is updated with cursor value

Test Coverage

• WebSocket test scenarios
• Protocol switching (host and member)
• Polling start/stop
• Event polling and processing
• Cursor management

Configuration

Default Values

• Polling interval: 2 seconds (from server pollingIntervalSeconds)
• WebSocket test timeout: 3 seconds

Cost Optimization

• WebSocket Priority: Uses WebSocket when available (lower cost)
• Polling Fallback: Only uses polling when WebSocket is unavailable
• Cursor-based: Efficient pagination with cursor

Breaking Changes

None. This is a backward-compatible addition.

Related

• Parent PR: feat: implement polling fallback for event communication smalruby3-develop#15
• Related Issue: feat: WebSocket 接続失敗時のポーリングによるイベント通知 smalruby3-develop#14
• Server PR: feat: implement polling fallback for event communication mesh-v2#40

🤖 Generated with Claude Code