openclaw/extensions/matrix/doctor-contract-api.ts

import type { Dirent } from "node:fs";
// Matrix API module exposes the plugin public contract.
import fs from "node:fs/promises";
import path from "node:path";
import { normalizeAccountId } from "openclaw/plugin-sdk/account-id";
import {
  archiveLegacyStateSource,
  type PluginDoctorStateMigration,
} from "openclaw/plugin-sdk/runtime-doctor-migrations";
import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
import {
  requiresExplicitMatrixDefaultAccount,
  resolveMatrixDefaultOrOnlyAccountId,
} from "./src/account-selection.js";
import {
  hasMatrixSyncCacheStateInStore,
  openMatrixSyncCacheStoreOptions,
  readLegacyMatrixSyncCacheState,
  writeMatrixSyncCacheStateToStore,
  type MatrixSyncCacheRecord,
} from "./src/matrix/client/file-sync-store.js";
import {
  hasMatrixStorageMetaStateInStore,
  normalizeMatrixStorageMetadata,
  openMatrixStorageMetaStoreOptions,
  writeMatrixStorageMetaStateToStore,
  type MatrixStorageMetadata,
} from "./src/matrix/client/storage.js";
import {
  MATRIX_CREDENTIALS_MAX_ENTRIES,
  MATRIX_CREDENTIALS_NAMESPACE,
  isMatrixCredentialRevocation,
  matrixCredentialsStoreKey,
  normalizeMatrixStoredCredentials,
  type MatrixCredentialStateRecord,
  type MatrixStoredCredentialRecord,
} from "./src/matrix/credentials-state.js";
import { migrateLegacyMatrixIdbSnapshot } from "./src/matrix/crypto-snapshot-doctor.js";
import {
  MATRIX_IDB_SNAPSHOT_FILENAME,
  MATRIX_LEGACY_CRYPTO_MIGRATION_FILENAME,
  MATRIX_RECOVERY_KEY_FILENAME,
  hasMatrixLegacyCryptoMigrationStateInStore,
  hasMatrixRecoveryKeyStateInStore,
  openMatrixLegacyCryptoMigrationStoreOptions,
  openMatrixRecoveryKeyStoreOptions,
  readLegacyMatrixLegacyCryptoMigrationState,
  readLegacyMatrixRecoveryKeyState,
  writeMatrixLegacyCryptoMigrationStateToStore,
  writeMatrixRecoveryKeyStateToStore,
  type MatrixLegacyCryptoMigrationState,
} from "./src/matrix/crypto-state-store.js";
import {
  collectMatrixInboundDedupeSources,
  hasCompletedMatrixInboundDedupeMigration,
  importNewestInboundDedupeMarkers,
  MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME,
  readLegacyInboundDedupeJsonSource,
  readLegacyInboundDedupeSqliteSource,
  recordMatrixInboundDedupeMigrationCompletion,
  reserveMatrixInboundDedupeMigrationCompletion,
  retireLegacyInboundDedupeSqliteRows,
  verifyMatrixInboundDedupeSourcesRetired,
  type LegacyInboundDedupeMarker,
  type MatrixInboundDedupeMigrationIo,
} from "./src/matrix/monitor/inbound-dedupe-migration.js";
import type { MatrixStoredRecoveryKey } from "./src/matrix/sdk/types.js";
import { resolveMatrixCredentialsDir } from "./src/storage-paths.js";

export { normalizeCompatibilityConfig, legacyConfigRules } from "./src/doctor-contract.js";

const MATRIX_SYNC_CACHE_FILENAME = "bot-storage.json";
const MATRIX_STORAGE_META_FILENAME = "storage-meta.json";

type LegacyMatrixCredentialSource = {
  accountId: string | null;
  filePath: string;
};

async function collectLegacyMatrixCredentialSources(params: {
  config: Parameters<PluginDoctorStateMigration["migrateLegacyState"]>[0]["config"];
  env: NodeJS.ProcessEnv;
  stateDir: string;
}): Promise<LegacyMatrixCredentialSource[]> {
  const credentialsDir = resolveMatrixCredentialsDir(params.stateDir);
  let entries: Dirent[];
  try {
    entries = await fs.readdir(credentialsDir, { withFileTypes: true });
  } catch {
    return [];
  }
  const files = entries
    .filter((entry) => entry.isFile() && /^credentials(?:-[a-z0-9._-]+)?\.json$/iu.test(entry.name))
    .toSorted((left, right) => {
      if (left.name === "credentials.json") {
        return 1;
      }
      if (right.name === "credentials.json") {
        return -1;
      }
      return left.name.localeCompare(right.name);
    });
  return files.map((entry) => {
    const match = /^credentials(?:-([a-z0-9._-]+))?\.json$/iu.exec(entry.name);
    const namedAccount = match?.[1];
    const accountId = namedAccount
      ? normalizeAccountId(namedAccount)
      : requiresExplicitMatrixDefaultAccount(params.config, params.env)
        ? null
        : normalizeAccountId(resolveMatrixDefaultOrOnlyAccountId(params.config, params.env));
    return { accountId, filePath: path.join(credentialsDir, entry.name) };
  });
}

async function readLegacyMatrixCredentials(
  source: LegacyMatrixCredentialSource,
): Promise<MatrixStoredCredentialRecord | null> {
  if (!source.accountId) {
    return null;
  }
  try {
    const raw = JSON.parse(await fs.readFile(source.filePath, "utf8")) as unknown;
    const createdAt =
      isRecord(raw) && typeof raw.createdAt === "string" && raw.createdAt
        ? raw.createdAt
        : (await fs.stat(source.filePath)).mtime.toISOString();
    return normalizeMatrixStoredCredentials(
      isRecord(raw) ? { ...raw, createdAt } : raw,
      source.accountId,
    );
  } catch {
    return null;
  }
}

async function collectLegacyMatrixStateRoots(
  stateDir: string,
  filename: string,
  options?: { includeMatrixRoot?: boolean },
): Promise<string[]> {
  const matrixRoot = path.join(stateDir, "matrix");
  const roots: string[] = [];
  async function visit(dir: string): Promise<void> {
    let entries: Dirent[];
    try {
      entries = await fs.readdir(dir, { withFileTypes: true });
    } catch {
      return;
    }
    for (const entry of entries) {
      const entryPath = path.join(dir, entry.name);
      if (entry.isFile() && entry.name === filename) {
        roots.push(dir);
        continue;
      }
      if (entry.isDirectory()) {
        await visit(entryPath);
      }
    }
  }
  await visit(matrixRoot);
  return roots
    .filter((root) => options?.includeMatrixRoot || path.resolve(root) !== path.resolve(matrixRoot))
    .toSorted();
}

async function collectLegacySyncCacheRoots(stateDir: string): Promise<string[]> {
  return collectLegacyMatrixStateRoots(stateDir, MATRIX_SYNC_CACHE_FILENAME);
}

async function readLegacyMatrixStorageMetadata(
  storageRootDir: string,
): Promise<MatrixStorageMetadata | null> {
  try {
    return normalizeMatrixStorageMetadata(
      JSON.parse(
        await fs.readFile(path.join(storageRootDir, MATRIX_STORAGE_META_FILENAME), "utf8"),
      ),
    );
  } catch {
    return null;
  }
}

async function archiveLegacySyncCache(params: {
  storageRootDir: string;
  changes: string[];
  warnings: string[];
  notices?: string[];
  notice?: string;
}): Promise<void> {
  await archiveLegacyMatrixStateFile({
    ...params,
    filename: MATRIX_SYNC_CACHE_FILENAME,
    label: "Matrix sync cache",
  });
}

async function archiveLegacyMatrixStateFile(params: {
  storageRootDir: string;
  filename: string;
  label: string;
  changes: string[];
  warnings: string[];
  notices?: string[];
  notice?: string;
}): Promise<void> {
  const warningCount = params.warnings.length;
  await archiveLegacyStateSource({
    filePath: path.join(params.storageRootDir, params.filename),
    label: params.label,
    changes: params.changes,
    warnings: params.warnings,
  });
  if (params.notice && params.warnings.length === warningCount) {
    params.notices?.push(params.notice);
  }
}

export const stateMigrations: PluginDoctorStateMigration[] = [
  {
    id: "matrix-credentials-json-to-plugin-state",
    label: "Matrix credentials",
    async detectLegacyState(params) {
      const sources = await collectLegacyMatrixCredentialSources(params);
      return sources.length > 0
        ? {
            preview: [
              `Matrix credential JSON can migrate to SQLite (${sources.length} ${sources.length === 1 ? "file" : "files"})`,
            ],
          }
        : null;
    },
    async migrateLegacyState(params) {
      const changes: string[] = [];
      const warnings: string[] = [];
      const sources = await collectLegacyMatrixCredentialSources(params);
      const store = params.context.openPluginStateKeyedStore<MatrixCredentialStateRecord>({
        namespace: MATRIX_CREDENTIALS_NAMESPACE,
        maxEntries: MATRIX_CREDENTIALS_MAX_ENTRIES,
        overflowPolicy: "reject-new",
      });
      for (const source of sources) {
        if (!source.accountId) {
          warnings.push(
            `Left ambiguous Matrix credential legacy source in place because no default account is selected: ${source.filePath}`,
          );
          continue;
        }
        const credentials = await readLegacyMatrixCredentials(source);
        if (!credentials) {
          warnings.push(
            `Left invalid Matrix credential legacy source in place: ${source.filePath}`,
          );
          continue;
        }
        const key = matrixCredentialsStoreKey(source.accountId);
        const stored = await store.lookup(key);
        if (isMatrixCredentialRevocation(stored, source.accountId)) {
          changes.push(
            `Archived revoked Matrix credential legacy source for account ${source.accountId}`,
          );
          await archiveLegacyStateSource({
            filePath: source.filePath,
            label: "Matrix credentials",
            changes,
            warnings,
          });
          continue;
        }
        const existing = normalizeMatrixStoredCredentials(stored, source.accountId);
        if (existing && JSON.stringify(existing) !== JSON.stringify(credentials)) {
          warnings.push(
            `Kept existing Matrix credentials for account ${source.accountId}; left differing legacy source in place`,
          );
          continue;
        }
        if (!existing) {
          try {
            await store.registerIfAbsent(key, credentials);
          } catch (error) {
            warnings.push(
              `Failed importing Matrix credentials for account ${source.accountId}: ${String(error)}; left legacy source in place`,
            );
            continue;
          }
        }
        const persisted = normalizeMatrixStoredCredentials(
          await store.lookup(key),
          source.accountId,
        );
        if (!persisted || JSON.stringify(persisted) !== JSON.stringify(credentials)) {
          warnings.push(
            `Failed verifying Matrix credentials for account ${source.accountId}; left legacy source in place`,
          );
          continue;
        }
        changes.push(`Migrated Matrix credentials for account ${source.accountId} to SQLite`);
        await archiveLegacyStateSource({
          filePath: source.filePath,
          label: "Matrix credentials",
          changes,
          warnings,
        });
      }
      return { changes, warnings };
    },
  },
  {
    id: "matrix-inbound-dedupe-to-claimable-dedupe",
    label: "Matrix inbound dedupe markers",
    async detectLegacyState(params) {
      return (await hasCompletedMatrixInboundDedupeMigration(params.context, params.env))
        ? null
        : { preview: ["Matrix inbound dedupe legacy sources need a one-time migration scan"] };
    },
    async migrateLegacyState(params) {
      const io: MatrixInboundDedupeMigrationIo = { context: params.context, env: params.env };
      const changes: string[] = [];
      const warnings: string[] = [];
      if (await hasCompletedMatrixInboundDedupeMigration(params.context, params.env)) {
        return { changes, warnings };
      }
      try {
        await reserveMatrixInboundDedupeMigrationCompletion(params.context, params.env);
      } catch (err) {
        warnings.push(
          `Failed reserving Matrix inbound dedupe migration completion: ${String(err)}; left legacy sources in place`,
        );
        return { changes, warnings };
      }
      const sources = await collectMatrixInboundDedupeSources(params.stateDir);
      if (sources.status === "incomplete") {
        warnings.push(...sources.warnings);
      }

      const recordCompletionIfClean = async (verifyRetirement = false) => {
        if (warnings.length > 0) {
          return;
        }
        if (verifyRetirement) {
          warnings.push(...(await verifyMatrixInboundDedupeSourcesRetired(params.stateDir)));
          if (warnings.length > 0) {
            return;
          }
        }
        try {
          await recordMatrixInboundDedupeMigrationCompletion(params.context, params.env);
          changes.push(
            `Recorded Matrix inbound dedupe migration completion (${sources.sqliteRoots.length} SQLite roots, ${sources.jsonRoots.length} JSON roots scanned)`,
          );
        } catch (err) {
          warnings.push(
            `Failed recording Matrix inbound dedupe migration completion: ${String(err)}`,
          );
        }
      };

      // Gather every marker first so the capacity-aware import keeps the
      // globally newest ones instead of whichever storage root imports last.
      const gathered: LegacyInboundDedupeMarker[] = [];
      const sqliteRootsToRetire: string[] = [];
      for (const storageRootDir of sources.sqliteRoots) {
        try {
          const source = await readLegacyInboundDedupeSqliteSource(storageRootDir);
          if (source.legacyRowCount === 0) {
            continue;
          }
          gathered.push(...source.markers);
          sqliteRootsToRetire.push(storageRootDir);
        } catch (err) {
          warnings.push(
            `Failed reading Matrix inbound dedupe rows for ${storageRootDir}: ${String(err)}; left legacy rows in place`,
          );
        }
      }
      const jsonRootsToRetire: string[] = [];
      for (const storageRootDir of sources.jsonRoots) {
        try {
          const markers = await readLegacyInboundDedupeJsonSource(storageRootDir);
          if (markers === null) {
            // Nothing recoverable, but archiving (rename, not delete) resolves
            // the pending detection while preserving the bytes for inspection.
            warnings.push(
              `Matrix inbound dedupe JSON for ${storageRootDir} is malformed; archived without import`,
            );
          } else {
            gathered.push(...markers);
          }
          jsonRootsToRetire.push(storageRootDir);
        } catch (err) {
          warnings.push(
            `Failed reading Matrix inbound dedupe JSON for ${storageRootDir}: ${String(err)}; left legacy file in place`,
          );
        }
      }
      if (sqliteRootsToRetire.length + jsonRootsToRetire.length === 0) {
        await recordCompletionIfClean();
        return { changes, warnings };
      }

      try {
        const result = await importNewestInboundDedupeMarkers({ io, markers: gathered });
        changes.push(
          `Migrated Matrix inbound dedupe markers to the claimable dedupe store (${result.imported} of ${result.total} entries)`,
        );
      } catch (err) {
        warnings.push(
          `Failed importing Matrix inbound dedupe markers: ${String(err)}; left legacy sources in place`,
        );
        return { changes, warnings };
      }

      // Retire the legacy sources only after the import succeeded so a failed
      // run keeps them for the next doctor attempt.
      for (const storageRootDir of sqliteRootsToRetire) {
        try {
          await retireLegacyInboundDedupeSqliteRows(storageRootDir);
          changes.push(`Retired Matrix inbound dedupe rows for ${storageRootDir}`);
        } catch (err) {
          warnings.push(
            `Failed retiring Matrix inbound dedupe rows for ${storageRootDir}: ${String(err)}`,
          );
        }
      }
      for (const storageRootDir of jsonRootsToRetire) {
        await archiveLegacyMatrixStateFile({
          storageRootDir,
          filename: MATRIX_LEGACY_INBOUND_DEDUPE_FILENAME,
          label: "Matrix inbound dedupe",
          changes,
          warnings,
        });
      }
      await recordCompletionIfClean(true);
      return { changes, warnings };
    },
  },
  {
    id: "matrix-storage-meta-json-to-plugin-state",
    label: "Matrix storage metadata",
    async detectLegacyState(params) {
      const previews: string[] = [];
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_STORAGE_META_FILENAME,
      )) {
        if (!(await readLegacyMatrixStorageMetadata(storageRootDir))) {
          continue;
        }
        previews.push(`Matrix storage metadata JSON can migrate to SQLite: ${storageRootDir}`);
      }
      return previews.length > 0 ? { preview: previews } : null;
    },
    async migrateLegacyState(params) {
      const changes: string[] = [];
      const warnings: string[] = [];
      const notices: string[] = [];
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_STORAGE_META_FILENAME,
      )) {
        const payload = await readLegacyMatrixStorageMetadata(storageRootDir);
        if (!payload) {
          continue;
        }
        const store = params.context.openPluginStateKeyedStore<MatrixStorageMetadata>(
          openMatrixStorageMetaStoreOptions(storageRootDir),
        );
        if (await hasMatrixStorageMetaStateInStore({ store })) {
          await archiveLegacyMatrixStateFile({
            storageRootDir,
            filename: MATRIX_STORAGE_META_FILENAME,
            label: "Matrix storage metadata",
            changes,
            warnings,
            notices,
            notice: `Kept existing Matrix storage metadata in SQLite and archived the legacy source for ${storageRootDir}`,
          });
          continue;
        }
        await writeMatrixStorageMetaStateToStore({ payload, store });
        changes.push(`Migrated Matrix storage metadata JSON to SQLite for ${storageRootDir}`);
        await archiveLegacyMatrixStateFile({
          storageRootDir,
          filename: MATRIX_STORAGE_META_FILENAME,
          label: "Matrix storage metadata",
          changes,
          warnings,
        });
      }
      return { changes, warnings, ...(notices.length > 0 ? { notices } : {}) };
    },
  },
  {
    id: "matrix-sync-cache-json-to-plugin-state",
    label: "Matrix sync cache",
    async detectLegacyState(params) {
      const previews: string[] = [];
      for (const storageRootDir of await collectLegacySyncCacheRoots(params.stateDir)) {
        const persisted = await readLegacyMatrixSyncCacheState(storageRootDir);
        if (!persisted) {
          continue;
        }
        previews.push(`Matrix sync cache JSON can migrate to SQLite: ${storageRootDir}`);
      }
      return previews.length > 0 ? { preview: previews } : null;
    },
    async migrateLegacyState(params) {
      const changes: string[] = [];
      const warnings: string[] = [];
      const notices: string[] = [];
      for (const storageRootDir of await collectLegacySyncCacheRoots(params.stateDir)) {
        const persisted = await readLegacyMatrixSyncCacheState(storageRootDir);
        if (!persisted) {
          continue;
        }
        const store = params.context.openPluginStateKeyedStore<MatrixSyncCacheRecord>(
          openMatrixSyncCacheStoreOptions(storageRootDir),
        );
        if (await hasMatrixSyncCacheStateInStore({ storageRootDir, store })) {
          await archiveLegacySyncCache({
            storageRootDir,
            changes,
            warnings,
            notices,
            notice: `Kept existing Matrix sync cache in SQLite and archived the legacy source for ${storageRootDir}`,
          });
          continue;
        }
        await writeMatrixSyncCacheStateToStore({
          storageRootDir,
          payload: persisted,
          store,
        });
        changes.push(`Migrated Matrix sync cache JSON to SQLite for ${storageRootDir}`);
        await archiveLegacySyncCache({ storageRootDir, changes, warnings });
      }
      return { changes, warnings, ...(notices.length > 0 ? { notices } : {}) };
    },
  },
  {
    id: "matrix-recovery-key-json-to-plugin-state",
    label: "Matrix recovery key",
    async detectLegacyState(params) {
      const previews: string[] = [];
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_RECOVERY_KEY_FILENAME,
      )) {
        if (!readLegacyMatrixRecoveryKeyState(storageRootDir)) {
          continue;
        }
        previews.push(`Matrix recovery-key JSON can migrate to SQLite: ${storageRootDir}`);
      }
      return previews.length > 0 ? { preview: previews } : null;
    },
    async migrateLegacyState(params) {
      const changes: string[] = [];
      const warnings: string[] = [];
      const notices: string[] = [];
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_RECOVERY_KEY_FILENAME,
      )) {
        const payload = readLegacyMatrixRecoveryKeyState(storageRootDir);
        if (!payload) {
          continue;
        }
        const store = params.context.openPluginStateKeyedStore<MatrixStoredRecoveryKey>(
          openMatrixRecoveryKeyStoreOptions(storageRootDir),
        );
        if (await hasMatrixRecoveryKeyStateInStore({ store })) {
          await archiveLegacyMatrixStateFile({
            storageRootDir,
            filename: MATRIX_RECOVERY_KEY_FILENAME,
            label: "Matrix recovery key",
            changes,
            warnings,
            notices,
            notice: `Kept existing Matrix recovery key in SQLite and archived the legacy source for ${storageRootDir}`,
          });
          continue;
        }
        await writeMatrixRecoveryKeyStateToStore({ payload, store });
        changes.push(`Migrated Matrix recovery-key JSON to SQLite for ${storageRootDir}`);
        await archiveLegacyMatrixStateFile({
          storageRootDir,
          filename: MATRIX_RECOVERY_KEY_FILENAME,
          label: "Matrix recovery key",
          changes,
          warnings,
        });
      }
      return { changes, warnings, ...(notices.length > 0 ? { notices } : {}) };
    },
  },
  {
    id: "matrix-legacy-crypto-migration-json-to-plugin-state",
    label: "Matrix legacy crypto state",
    async detectLegacyState(params) {
      const previews: string[] = [];
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_LEGACY_CRYPTO_MIGRATION_FILENAME,
        { includeMatrixRoot: true },
      )) {
        if (!readLegacyMatrixLegacyCryptoMigrationState(storageRootDir)) {
          continue;
        }
        previews.push(
          `Matrix legacy crypto migration JSON can migrate to SQLite: ${storageRootDir}`,
        );
      }
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_IDB_SNAPSHOT_FILENAME,
        { includeMatrixRoot: true },
      )) {
        previews.push(`Matrix IndexedDB snapshot JSON can migrate to SQLite: ${storageRootDir}`);
      }
      return previews.length > 0 ? { preview: previews } : null;
    },
    async migrateLegacyState(params) {
      const changes: string[] = [];
      const warnings: string[] = [];
      const notices: string[] = [];
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_LEGACY_CRYPTO_MIGRATION_FILENAME,
        { includeMatrixRoot: true },
      )) {
        const state = readLegacyMatrixLegacyCryptoMigrationState(storageRootDir);
        if (!state) {
          continue;
        }
        const store = params.context.openPluginStateKeyedStore<MatrixLegacyCryptoMigrationState>(
          openMatrixLegacyCryptoMigrationStoreOptions(storageRootDir),
        );
        if (await hasMatrixLegacyCryptoMigrationStateInStore({ store })) {
          await archiveLegacyMatrixStateFile({
            storageRootDir,
            filename: MATRIX_LEGACY_CRYPTO_MIGRATION_FILENAME,
            label: "Matrix legacy crypto migration",
            changes,
            warnings,
            notices,
            notice: `Kept existing Matrix legacy crypto migration in SQLite and archived the legacy source for ${storageRootDir}`,
          });
          continue;
        }
        await writeMatrixLegacyCryptoMigrationStateToStore({ state, store });
        changes.push(
          `Migrated Matrix legacy crypto migration JSON to SQLite for ${storageRootDir}`,
        );
        await archiveLegacyMatrixStateFile({
          storageRootDir,
          filename: MATRIX_LEGACY_CRYPTO_MIGRATION_FILENAME,
          label: "Matrix legacy crypto migration",
          changes,
          warnings,
        });
      }
      for (const storageRootDir of await collectLegacyMatrixStateRoots(
        params.stateDir,
        MATRIX_IDB_SNAPSHOT_FILENAME,
        { includeMatrixRoot: true },
      )) {
        await migrateLegacyMatrixIdbSnapshot({
          storageRootDir,
          context: params.context,
          changes,
          notices,
          warnings,
        });
      }
      return { changes, warnings, ...(notices.length > 0 ? { notices } : {}) };
    },
  },
];