Class MastraStateAdapter

packages/core/src/channels/state-adapter.ts:19–244

 * short-lived (seconds to minutes) and don't need persistence.
 */
export class MastraStateAdapter implements StateAdapter {
  private memoryStore: MemoryStorage;
  private connected = false;
  private connectPromise: Promise<void> | null = null;

  // In-memory ephemeral state (cache, locks, lists, queues)
  private readonly cache = new Map<string, CachedValue>();
  private readonly locks = new Map<string, Lock>();
  private readonly lists = new Map<string, { values: unknown[]; expiresAt: number | null }>();
  private readonly queues = new Map<string, QueueEntry[]>();

  constructor(memoryStore: MemoryStorage) {
    this.memoryStore = memoryStore;
  }

  async connect(): Promise<void> {
    if (this.connected) return;
    if (!this.connectPromise) {
      this.connectPromise = Promise.resolve().then(() => {
        this.connected = true;
      });
    }
    await this.connectPromise;
  }

  async disconnect(): Promise<void> {
    this.connected = false;
    this.connectPromise = null;
    this.cache.clear();
    this.locks.clear();
    this.lists.clear();
    this.queues.clear();
  }

  // ---------------------------------------------------------------------------
  // Subscriptions — persisted via Mastra thread metadata
  // ---------------------------------------------------------------------------

  async subscribe(threadId: string): Promise<void> {
    // Find the Mastra thread mapped to this external thread ID and mark it
    const thread = await this.findThreadByExternalId(threadId);
    if (!thread) return; // Thread not yet mapped — subscribe will be a no-op
    await this.memoryStore.updateThread({
      id: thread.id,
      title: thread.title ?? '',
      metadata: { ...thread.metadata, channel_subscribed: 'true' },
    });
  }

  async unsubscribe(threadId: string): Promise<void> {
    const thread = await this.findThreadByExternalId(threadId);
    if (!thread) return;
    await this.memoryStore.updateThread({
      id: thread.id,
      title: thread.title ?? '',
      metadata: { ...((thread.metadata ?? {}) as Record<string, unknown>), channel_subscribed: 'false' },
    });
  }