/**
 * The public half of hosted-MCP intake (TASK-172, scope §4): the Client ID
 * Metadata Document, and the start route that sends a member to their vendor's
 * consent screen.
 *
 * Mounted under `/api/integrations/connect/hosted-mcp` by
 * `routes/integrations.ts`, on a path deeper than `/:podId` so an entry id can
 * never be read as a pod id.
 *
 * The document is unauthenticated on purpose and holds nothing to protect: the
 * authorization server fetches it before any member has consented, and it is
 * the same document for every member of the instance (one client id per
 * instance per entry, §4). A per-person client would need registration per
 * person, which is the thing CIMD exists to avoid.
 */
const express = require('express');
import type { Request, Response } from 'express';
import type { HostedMcpAuthorizationServer } from '../services/hostedMcpIntakeService';
import type { HostedMcpEntry } from '../services/hostedMcpEntryService';
import rateLimit from 'express-rate-limit';
// eslint-disable-next-line global-require
const auth = require('../middleware/auth');
import { writeIntegrationsRateLimit } from '../middleware/integrationRateLimit';
import { cloudflareIpRateLimitKeyGenerator } from '../middleware/ipRateLimit';
import { upstreamFetch } from '../services/upstreamFetch';
import { dcrClientForRow, ensureDcrClient, refuseDcrClient } from '../services/hostedMcpDcrRegistry';
import { forgetHostedMcpDrift } from '../services/hostedMcpToolDefinitions';
// eslint-disable-next-line global-require
const Integration = require('../models/Integration');
// eslint-disable-next-line global-require
const connectorSecrets = require('../services/connectorSecrets');
const { secretWriteGeneration } = require('../services/hostedMcpSecretGeneration');
// eslint-disable-next-line global-require
const { HOSTED_MCP_ACCESS_TOKEN, HOSTED_MCP_REFRESH_TOKEN } = require('../services/connectorSecretKinds');
// eslint-disable-next-line global-require
const { revokeConnectionGrants } = require('../services/roomGrantService');
// eslint-disable-next-line global-require
const { revokeTokenAtVendor } = require('../services/connectionRemovalService');
// eslint-disable-next-line global-require
const { HOSTED_MCP_ENTRIES, findHostedMcpEntry, hostedMcpRevokeTarget } = require('../integrations/hostedMcp/entries');
const { isHostedMcpEntryHidden } = require('../integrations/hostedMcp/visibility');
// eslint-disable-next-line global-require
const {
  HOSTED_MCP_NONCE_COOKIE,
  HOSTED_MCP_PENDING_TTL_MS,
  browserNonceMatches,
  buildAuthorizeUrl,
  buildClientMetadataDocument,
  buildTokenExchangeRequest,
  createBrowserNonce,
  createPkcePair,
  createStateNonce,
  discoverAuthorizationServer,
  hostedMcpCallbackUrl,
  hostedMcpNonceCookiePath,
  readCookie,
  resolvedHostedMcpClient,
} = require('../services/hostedMcpIntakeService');

type AuthedRequest = Request & { user?: { id?: string; role?: string } };

/** A vendor-side failure is not the caller's fault and not a 4xx. */
const vendorFailureStatus = (code?: string): number => (
  code === 'issuer_unreachable' || code === 'issuer_metadata_incomplete' || code === 'issuer_mismatch'
    ? 502
    : 503
);

/**
 * The browser comes back here, so every outcome is a redirect and not JSON:
 * a person who refused consent has to land on the page that can offer Connect
 * again. Mirrors the Slack callback's `?slack=` contract in shape.
 */
const publicAppUrl = (): string => {
  const configured = String(process.env.PUBLIC_APP_URL || process.env.FRONTEND_URL || 'https://commonly.me')
    .split(',')[0]
    .trim();
  return (configured || 'https://commonly.me').replace(/\/+$/, '');
};

const callbackRedirect = (
  res: Response,
  status: string,
  code?: string,
  options?: {
    manualRevoke?: { entryId: string; revokeAt: string };
  },
) => {
  const query = new URLSearchParams({ hostedMcp: status });
  if (code) query.set('code', code);
  if (options?.manualRevoke) {
    query.set('entryId', options.manualRevoke.entryId);
    query.set('revokeAt', options.manualRevoke.revokeAt);
  }
  return res.redirect(302, `${publicAppUrl()}/v2/connectors?${query.toString()}`);
};

/** Vendors return OAuth scope lists with either separator; compare the set. */
const parseScopeSet = (scope: string): Set<string> => new Set(
  scope.split(/[\s,]+/).map((value) => value.trim()).filter(Boolean),
);

/** A refused grant is never stored; revoke its refresh token when the entry can. */
const revokeRefusedGrant = async (
  entry: HostedMcpEntry,
  client: { clientId: string; clientSecret?: string },
  tokens: TokenResponse,
): Promise<void> => {
  if (!hostedMcpRevokeTarget(entry)?.endpoint) return;
  const token = tokens.refresh_token || tokens.access_token;
  if (!token) return;
  try {
    await revokeTokenAtVendor({
      entry,
      clientId: client.clientId,
      ...(client.clientSecret ? { clientSecret: client.clientSecret } : {}),
      token,
      tokenTypeHint: tokens.refresh_token ? 'refresh_token' : 'access_token',
    });
  } catch {
    // The connect remains refused and no token is retained if vendor revoke fails.
  }
};

/**
 * `sub` from an ID token the token endpoint handed us over TLS in answer to our
 * own exchange. The signature is not verified because there is no third party
 * here to defend against: we are not accepting this token from anyone, and the
 * claims only need to be as trustworthy as the response they arrived in. This
 * is the only source of `providerSubject`, §2's account-change key, so a
 * vendor that issues no ID token leaves it undefined and §2's comparison falls
 * to the "cannot be compared" arm.
 */
const idTokenSubject = (idToken?: string): string | undefined => {
  const payload = typeof idToken === 'string' ? idToken.split('.')[1] : undefined;
  if (!payload) return undefined;
  try {
    const claims = JSON.parse(Buffer.from(payload, 'base64url').toString('utf8')) as { sub?: unknown };
    return typeof claims.sub === 'string' && claims.sub ? claims.sub : undefined;
  } catch {
    return undefined;
  }
};

type TokenResponse = {
  access_token?: string;
  refresh_token?: string;
  expires_in?: number;
  scope?: string;
  id_token?: string;
};

const router: ReturnType<typeof express.Router> = express.Router();

// Both GETs below are public and unauthenticated on purpose: `client-metadata`
// is fetched by the authorization server before any member has consented, and
// `callback` is a browser navigation from the vendor, where the httpOnly cookie
// is the flow binding rather than an authorization. `routeRateLimitGuard` and
// CodeQL's js/missing-rate-limiting both require a limiter AHEAD of auth on
// every registration, and here there is no auth at all — so this limiter is the
// only thing in front of an unauthenticated row lookup and token exchange.
//
// Keyed by the Cloudflare-aware IP rule (`middleware/ipRateLimit.ts:49`),
// because `req.ip` walks to a cloudflared pod address for every external caller
// — one bucket for the whole internet (TASK-110's class).
//
// Two limiter instances, not one shared: a client that hammers the static
// document must not be able to spend the callback's allowance, which is the one
// that finishes a member's connect.
//
// The two BOUNDS are deliberately different, because the two callers are, and
// the guard above cannot check that — it reads names, so any number satisfies it.
// So are the two REFUSALS: this route file's contract is that every callback
// outcome is a redirect to the page, a contract a shared JSON handler breaks on
// exactly the one path where the caller is a person mid-navigation.
const PUBLIC_GET_WINDOW_MS = 60_000;
const publicGetRateLimit = (
  max: number,
  label: string,
  refuse: (res: Response, detail: { max: number, label: string }) => void,
) => rateLimit({
  windowMs: PUBLIC_GET_WINDOW_MS,
  max,
  standardHeaders: true,
  legacyHeaders: false,
  keyGenerator: cloudflareIpRateLimitKeyGenerator,
  handler: (_req: unknown, res: Response) => refuse(res, { max, label }),
});

// `client-metadata`'s caller is a server, so its refusal is machine-readable.
const jsonRefusal = (res: Response, detail: { max: number; label: string }) => {
  res.status(429).json({ msg: `rate limit exceeded: ${detail.max} requests per 60s (${detail.label})` });
};

// `callback`'s caller is a browser mid-navigation, and a throttled person is the
// one least able to work out why the address bar says JSON. The refusal is the
// same redirect the rest of the route uses, carrying a code the page can speak
// to rather than a body only a developer would read.
const browserRefusal = (res: Response) => {
  callbackRedirect(res, 'error', 'rate_limited');
};

// Fetched by the AUTHORIZATION SERVER, not by a person, and it is one client id
// per instance per entry (§4) — so a vendor's handful of egress addresses fetch
// this document on behalf of every member. A bound tuned for humans would
// surface at the vendor as a client whose metadata document cannot be read, i.e.
// as a refused authorization for everyone at once, with nothing in our own logs
// pointing at a limiter. Loose on purpose: the document is static, identical for
// every member, and holds no secret.
const clientMetadataRateLimit = publicGetRateLimit(600, 'client-metadata', jsonRefusal);

// A browser, once per connect. 60 a minute is far above a human retry pattern
// and still bounds the route that validates `state` and the browser nonce and
// then performs a token exchange.
const callbackRateLimit = publicGetRateLimit(60, 'callback', browserRefusal);

router.get('/:entryId/client-metadata', clientMetadataRateLimit, (req: Request, res: Response) => {
  const entry = findHostedMcpEntry(HOSTED_MCP_ENTRIES, req.params.entryId);
  if (!entry) {
    // An unknown entry is a 404 and not a document with empty fields: a
    // document that names no redirect URI is one an authorization server could
    // still accept as a client with nowhere to send a code.
    return res.status(404).json({ error: 'unknown_entry' });
  }
  if (entry.client !== 'cimd') {
    return res.status(404).json({ error: 'client_metadata_not_used' });
  }
  return res.json(buildClientMetadataDocument(entry));
});

/**
 * Begin a connect. Discovery runs BEFORE the row is written, so a vendor outage
 * leaves no pending row behind — a row whose only content is a dead nonce is a
 * connect the page would show as in progress and nothing would ever finish.
 *
 * The row is found by `(createdBy, config.entryId)` in the filter rather than
 * inserted blind, because §2 allows exactly one row per person per entry: a
 * second Connect reuses the row's pending state and the callback replaces the
 * pair through the §10.3 fence. An insert here would collide with the partial
 * unique index instead.
 */
const startConnect = async (req: AuthedRequest, res: Response) => {
  const userId = req.user?.id;
  if (!userId) {
    return res.status(401).json({ error: 'authentication_required' });
  }
  const entry = findHostedMcpEntry(HOSTED_MCP_ENTRIES, req.params.entryId);
  if (!entry) {
    return res.status(404).json({ error: 'unknown_entry' });
  }
  if (isHostedMcpEntryHidden(entry.id, process.env, userId)) return res.status(404).json({ error: 'unknown_entry' });

  let client: { clientId: string; clientSecret?: string } | null = null;
  if (entry.client !== 'dcr') {
    try {
      client = resolvedHostedMcpClient(entry);
    } catch (error) {
      const code = (error as { code?: string }).code;
      return res.status(vendorFailureStatus(code)).json({ error: code || 'client_not_configured' });
    }
  }

  let authorizationServer: HostedMcpAuthorizationServer;
  try {
    authorizationServer = await discoverAuthorizationServer(entry.issuer);
  } catch (error) {
    const code = (error as { code?: string }).code;
    return res.status(vendorFailureStatus(code)).json({ error: code });
  }

  try {
    if (entry.client === 'dcr') {
      client = await ensureDcrClient(entry, authorizationServer, hostedMcpCallbackUrl(entry.id));
    }
  } catch (error) {
    const code = (error as { code?: string }).code;
    return res.status(vendorFailureStatus(code)).json({ error: code || 'client_registration_refused' });
  }
  if (!client) return res.status(503).json({ error: 'client_not_configured' });

  const { verifier, challenge } = createPkcePair();
  const state = createStateNonce();
  // Two secrets, and the asymmetry between them is the point: `state` proves
  // the flow EXISTS and travels in a URL, the nonce proves the browser that
  // FINISHES it is the one that started it and travels only as a cookie.
  const browserNonce = createBrowserNonce();
  const expiresAt = new Date(Date.now() + HOSTED_MCP_PENDING_TTL_MS);
  const redirectUri = hostedMcpCallbackUrl(entry.id);

  // `intake` is written here as well as at the callback: it is the one field
  // that says which flow owns the pending state, and a row left `pending` with
  // no nonce would otherwise be indistinguishable from one whose nonce expired.
  // Legacy connected rows predate credentialAuthGeneration. Bind their existing
  // pair to the generation that published it BEFORE /start advances the pending
  // authorization. Refresh admission is fenced while pendingAuth is present,
  // so no caller can slip between this CAS and the new flow's write.
  const legacyCredential = await Integration.findOne({
    type: 'hosted-mcp', createdBy: userId, 'config.entryId': entry.id,
    status: 'connected', 'config.credentialAuthGeneration': { $exists: false },
    $or: [
      { 'config.credentialRef': { $exists: true } },
      { 'config.refreshTokenRef': { $exists: true } },
    ],
  }).select('_id config.authGeneration').lean();
  if (legacyCredential) {
    await Integration.updateOne(
      { _id: legacyCredential._id, 'config.credentialAuthGeneration': { $exists: false } },
      { $set: { 'config.credentialAuthGeneration': legacyCredential.config?.authGeneration ?? 0 } },
    );
  }
  const row = await Integration.findOneAndUpdate(
    { type: 'hosted-mcp', createdBy: userId, 'config.entryId': entry.id },
    {
      $inc: { 'config.authGeneration': 1 },
      $set: {
        'config.intake': 'oauth',
        'config.pendingAuth': { state, codeVerifier: verifier, expiresAt, browserNonce, clientId: client.clientId },
      },
      $setOnInsert: {
        type: 'hosted-mcp',
        scope: 'user',
        status: 'pending',
        createdBy: userId,
        'config.entryId': entry.id,
      },
    },
    { upsert: true, new: true, setDefaultsOnInsert: true },
  );

  // Scoped to THIS entry's callback, so it is not sent to any other route, and
  // kept out of JavaScript, because a nonce a script can read is a nonce a
  // script can hand to somebody else. The options stay at the sink — the same
  // shape as Slack's — so a reviewer reads them where the cookie is written.
  res.cookie?.(HOSTED_MCP_NONCE_COOKIE, browserNonce, {
    httpOnly: true,
    sameSite: 'lax',
    secure: true,
    maxAge: HOSTED_MCP_PENDING_TTL_MS,
    path: hostedMcpNonceCookiePath(entry.id),
  });

  return res.json({
    authorizeUrl: buildAuthorizeUrl(entry, authorizationServer.authorization_endpoint, {
      clientId: client.clientId, redirectUri, state, codeChallenge: challenge,
    }),
    expiresAt,
    integrationId: row?._id,
  });
};

router.post('/:entryId/start', writeIntegrationsRateLimit, auth, async (req: AuthedRequest, res: Response) => {
  try {
    await startConnect(req, res);
  } catch {
    console.error('[hosted-mcp] start failed', JSON.stringify({ entryId: req.params.entryId }));
    if (!res.headersSent) res.status(503).json({ error: 'connector_unavailable' });
  }
});

/**
 * The callback (§4). Everything a member consented to is spent by the time we
 * get here, so the state is consumed BEFORE the exchange: a state that could be
 * spent twice is a second consent nobody gave, and a vendor code replayed at
 * the token endpoint is a code the AS may have already retired.
 *
 * Two secrets guard it, and they fail differently (§10.7). The cookie is
 * REFUSED WITHOUT ANY ROW READ when it is absent, and refused without consuming
 * the flow when it is present but belongs to a different flow. Only then is the
 * state claimed.
 *
 * The claim is `$unset` of `pendingAuth` rather than the Slack callback's
 * `claimId`, and not by preference: `config.pendingAuth` is a strict subdocument
 * (`models/Integration.ts:375`), so a `claimId` written onto it is dropped
 * silently and no row would ever appear claimed. Consuming the whole subdocument
 * is atomic on the same terms and hands back the verifier in the pre-image,
 * which is the only place it is kept.
 */
interface CallbackFlow {
  pendingRowId?: string;
  connectedRowId?: string;
  credentialWriteStarted?: boolean;
  authGeneration?: number;
}

const finishCallback = async (req: Request, res: Response, flow: CallbackFlow) => {
  const entry = findHostedMcpEntry(HOSTED_MCP_ENTRIES, req.params.entryId);
  if (!entry) {
    return callbackRedirect(res, 'error', 'unknown_entry');
  }
  const state = typeof req.query.state === 'string' ? req.query.state : '';
  const code = typeof req.query.code === 'string' ? req.query.code : '';
  const vendorError = typeof req.query.error === 'string' ? req.query.error : '';
  if (vendorError) {
    // The provider denied authorization before issuing a code. Leave the
    // pending flow intact so the member can retry; never forward vendor text.
    return callbackRedirect(res, 'error', vendorError === 'access_denied' ? 'consent_denied' : 'vendor_error');
  }
  if (!state || !code) {
    return callbackRedirect(res, 'error', 'invalid_state');
  }
  // Presence first, and before the row read: a browser that never started a
  // flow must not be able to finish one by holding a state it was sent, and the
  // cheapest refusal is the one that touches nothing.
  const browserNonce = readCookie(req.headers.cookie, HOSTED_MCP_NONCE_COOKIE);
  if (!browserNonce) {
    return callbackRedirect(res, 'error', 'invalid_state');
  }
  // RFC 9207. Absent is not a refusal — a vendor that sends no `iss` still had
  // its code delivered to a redirect URI naming THIS entry — but a present and
  // different one is another authorization server's answer.
  const iss = typeof req.query.iss === 'string' ? req.query.iss : '';
  if (iss && iss !== entry.issuer) {
    return callbackRedirect(res, 'error', 'issuer_mismatch');
  }

  const now = new Date();
  const issued = await Integration.findOne({
    type: 'hosted-mcp', 'config.entryId': entry.id, 'config.pendingAuth.state': state,
  });
  if (!issued) {
    return callbackRedirect(res, 'error', 'invalid_state');
  }
  const expiresAt = issued.config?.pendingAuth?.expiresAt;
  if (!expiresAt || new Date(expiresAt).getTime() <= now.getTime()) {
    return callbackRedirect(res, 'error', 'state_expired');
  }
  // The state is unguessable, but the STARTER is who holds it: sending that URL
  // to somebody else is enough to have their vendor credential delivered onto
  // the starter's row, and every trail row would correctly name the starter.
  // This is the check that makes the browser finishing the flow the one that
  // started it. It sits OUTSIDE the claim below on purpose, so a wrong browser
  // is refused WITHOUT consuming a flow that belongs to somebody else.
  if (!browserNonceMatches(issued.config?.pendingAuth?.browserNonce, browserNonce)) {
    return callbackRedirect(res, 'error', 'browser_mismatch');
  }
  if (issued.status === 'pending') flow.pendingRowId = String(issued._id);

  const codeVerifier = issued.config?.pendingAuth?.codeVerifier;
  const consumed = await Integration.findOneAndUpdate(
    {
      _id: issued._id,
      'config.pendingAuth.state': state,
      'config.pendingAuth.expiresAt': { $gt: now },
    },
    { $unset: { 'config.pendingAuth': 1 } },
    { new: false },
  );
  if (!consumed || !codeVerifier) {
    // A second delivery of the same state, or two tabs finishing at once.
    return callbackRedirect(res, 'error', 'state_consumed');
  }
  flow.authGeneration = consumed.config?.authGeneration;
  if (consumed.status === 'connected') flow.connectedRowId = String(consumed._id);

  let authorizationServer: HostedMcpAuthorizationServer;
  try {
    authorizationServer = await discoverAuthorizationServer(entry.issuer);
  } catch (error) {
    return callbackRedirect(res, 'error', (error as { code?: string }).code || 'issuer_unreachable');
  }

  let client: { clientId: string; clientSecret?: string };
  try {
    if (entry.client === 'dcr') {
      const recorded = await dcrClientForRow(entry, consumed.config?.pendingAuth?.clientId);
      if (!recorded) return callbackRedirect(res, 'error', 'client_auth_refused');
      client = recorded;
    } else {
      client = resolvedHostedMcpClient(entry);
    }
  } catch (error) {
    return callbackRedirect(res, 'error', (error as { code?: string }).code || 'client_not_configured');
  }

  let tokens: TokenResponse;
  try {
    const tokenRequest = buildTokenExchangeRequest(entry, authorizationServer, client, {
      redirectUri: hostedMcpCallbackUrl(entry.id),
      code,
      codeVerifier,
    });
    const response = await upstreamFetch(authorizationServer.token_endpoint, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/x-www-form-urlencoded',
        Accept: 'application/json',
        ...tokenRequest.headers,
      },
      body: tokenRequest.body,
    });
    if (!response.ok) {
      const refusal = await response.json().catch(() => null) as { error?: unknown } | null;
      if (refusal?.error === 'invalid_client') {
        await refuseDcrClient(entry, client.clientId);
        return callbackRedirect(res, 'error', 'client_auth_refused');
      }
      return callbackRedirect(res, 'error', 'exchange_refused');
    }
    tokens = (await response.json()) as TokenResponse;
  } catch {
    return callbackRedirect(res, 'error', 'exchange_unreachable');
  }
  if (!tokens?.access_token) {
    return callbackRedirect(res, 'error', 'exchange_incomplete');
  }

  const scopeWasReturned = Object.prototype.hasOwnProperty.call(tokens, 'scope');
  if (scopeWasReturned && typeof tokens.scope !== 'string') {
    await revokeRefusedGrant(entry, client, tokens);
    return callbackRedirect(res, 'error', 'exchange_invalid_scope');
  }
  const grantedScope = scopeWasReturned ? tokens.scope as string : entry.scopes.join(' ');
  const grantedScopes: Set<string> = scopeWasReturned
    ? parseScopeSet(grantedScope)
    : new Set<string>(entry.scopes);
  const requestedScopes: Set<string> = new Set<string>(entry.scopes);
  const extraScopes: string[] = [...grantedScopes].filter((scope) => !requestedScopes.has(scope));
  if (extraScopes.length) {
    // Scope strings come from the provider and can contain credential material.
    // Log only the entry and count, never the values.
    console.warn('[hosted-mcp] refused overbroad OAuth scope grant', JSON.stringify({
      entryId: entry.id,
      extraScopeCount: extraScopes.length,
    }));
    await revokeRefusedGrant(entry, client, tokens);
    return callbackRedirect(res, 'error', 'unrequested_scope');
  }
  // Store the parsed set in the canonical OAuth space-delimited form. This
  // keeps a vendor's comma-separated response useful to later scope checks.
  const recordedGrantedScope = [...grantedScopes].join(' ');

  const owner = String(consumed.createdBy);
  // The credential half of the row lives under `config` (a strict
  // subdocument), so a top-level read is `undefined` on every real row: the
  // account-change rule would never fire and the retired refresh token would
  // never be dropped. The arms below read the same shape.
  const previousSubject = consumed.config?.providerSubject || undefined;
  const providerSubject = idTokenSubject(tokens.id_token);
  const hasPreviousCredential = Boolean(
    consumed.config?.credentialRef || consumed.config?.refreshTokenRef,
  );
  const subjectsComparable = Boolean(previousSubject && providerSubject);
  const subjectChanged = subjectsComparable && previousSubject !== providerSubject;
  // §2: a grant was made against the reach of the account connected at the
  // time, and the row's `createdAt` survives a reconnect, so the broker's
  // TASK-148 guard cannot see this one. An unknowable subject is not proof of a
  // same-account reconnect, so its grants are revoked and the person gets the
  // vendor page to decide whether an older authorization still needs removing.
  const accountChanged = hasPreviousCredential && (!subjectsComparable || subjectChanged);
  if (accountChanged) {
    await revokeConnectionGrants({ connection: consumed, revokedBy: owner });
  }

  let manualRevokeAt: string | undefined;
  if (hasPreviousCredential && !subjectsComparable) {
    // Never send an uncomparable token: it may be the same account that just
    // authorized the new pair, and some vendors revoke all grants for that
    // account/client together.
    manualRevokeAt = hostedMcpRevokeTarget(entry)?.page;
  } else if (hasPreviousCredential && subjectChanged) {
    const revokeTarget = hostedMcpRevokeTarget(entry);
    const recordedClientId = consumed.config?.clientId;
    const previousClient = entry.client === 'dcr'
      ? await dcrClientForRow(entry, recordedClientId)
      : (!recordedClientId || recordedClientId === client.clientId ? client : null);
    const clientMatches = Boolean(previousClient);
    const previousTokenRef = consumed.config?.refreshTokenRef || consumed.config?.credentialRef;
    let withdrawn = false;

    if (revokeTarget?.endpoint && clientMatches && previousTokenRef) {
      let previousToken: unknown;
      try {
        // This read must precede the first put below. connectorSecrets.put
        // replaces the ciphertext under this same kind/ref on reconnect.
        previousToken = await connectorSecrets.get(String(previousTokenRef));
      } catch {
        // The callback still finishes; the provider page is the only recovery
        // once the new token pair overwrites this one.
      }

      if (typeof previousToken === 'string' && previousToken) {
        try {
          await revokeTokenAtVendor({
            entry,
            clientId: previousClient!.clientId,
            ...(previousClient!.clientSecret ? { clientSecret: previousClient!.clientSecret } : {}),
            token: previousToken,
            tokenTypeHint: consumed.config?.refreshTokenRef ? 'refresh_token' : 'access_token',
          });
          withdrawn = true;
        } catch {
          // Do not stop the successful reconnect or expose provider details.
          // The old token cannot be retried after the secret ref is overwritten.
        }
      }
    }

    if (!withdrawn) manualRevokeAt = revokeTarget?.page;
  }

  // Fence publication before either stable secret ref can be replaced. A
  // newer /start increments authGeneration; its callback may then finish while
  // this one is waiting for the secret store. Both the row and each encrypted
  // secret reject an older generation, and the row is unusable during writes.
  const authGeneration = flow.authGeneration ?? 0;
  const generationFilter = flow.authGeneration === undefined
    ? { 'config.authGeneration': { $exists: false } }
    : { 'config.authGeneration': authGeneration };
  const retired = await Integration.findOneAndUpdate(
    { _id: consumed._id, ...generationFilter, 'config.pendingAuth': { $exists: false } },
    {
      $set: {
        status: 'error',
        errorMessage: 'Connection could not be completed. Try again.',
        errorMessageUserFacing: true,
      },
    },
    { new: true },
  );
  if (!retired) return callbackRedirect(res, 'error', 'state_consumed');
  // Clear the old-grant assessment before replacement begins.
  forgetHostedMcpDrift(String(consumed._id));
  flow.credentialWriteStarted = true;
  // A callback is refresh 0 of its own authorization: every refresh that the
  // previous authorization still has in flight ranks below it.
  const writeGeneration = secretWriteGeneration(authGeneration, 0);
  const credentialRef = await connectorSecrets.put(
    String(consumed._id), HOSTED_MCP_ACCESS_TOKEN, tokens.access_token, writeGeneration,
  );
  let refreshTokenRef = consumed.config?.refreshTokenRef || undefined;
  if (tokens.refresh_token) {
    refreshTokenRef = await connectorSecrets.put(
      String(consumed._id), HOSTED_MCP_REFRESH_TOKEN, tokens.refresh_token, writeGeneration,
    );
  } else if (refreshTokenRef) {
    // This authorization returned no refresh token, so the row cannot confirm
    // that its stored token belongs to this grant. Discard it rather than let a
    // later refresh use a credential whose grant binding is unknown.
    // Do not delete a stable ref here: a newer callback may have already
    // replaced it. Dropping the row reference lets the orphan sweep remove it.
    refreshTokenRef = undefined;
  }

  const published = await Integration.findOneAndUpdate(
    { _id: consumed._id, status: 'error', ...generationFilter, 'config.pendingAuth': { $exists: false } },
    {
      $set: {
        status: 'connected',
        // The id used on this exact exchange is the durable pair discriminator
        // (§2). Instance configuration remains the source used by later calls.
        'config.clientId': client.clientId,
        // Every one of these is a `config.*` path: `config` is a STRICT
        // subdocument, so an unprefixed `credentialRef` is dropped in silence
        // and this handler reports a Connection whose row holds no token (see
        // the model's note, and the arm that now reads the schema).
        // The page this entry names, copied onto the row because the entry
        // itself can leave the catalogue and the row must still be finishable
        // (ruling 75780; `connectionRemovalService.REVOKE_PAGE_KEY`). The
        // `endpoint` is deliberately NOT copied: a stored endpoint would be a
        // URL a row carries into a POST, and the entry is the only thing that
        // decides where a token goes.
        'config.revokePage': hostedMcpRevokeTarget(entry)?.page || null,
        'config.credentialRef': credentialRef,
        'config.refreshTokenRef': refreshTokenRef || null,
        'config.refreshGeneration': 0,
        'config.credentialAuthGeneration': authGeneration,
        'config.grantedScope': recordedGrantedScope,
        'config.expiresAt': tokens.expires_in ? new Date(Date.now() + tokens.expires_in * 1000) : null,
        // `|| null`, not the bare value: `$set` with `undefined` is a NO-OP on
        // this strict subdocument (measured on a real mongod — the old value
        // survives), and a surviving subject is a durable lie about which
        // vendor account is connected. The next reconnect would compare
        // against it, read "same account", and keep grants minted under the
        // account that held the row in between. Cleared, the comparison is two
        // unknowns and revokes, which is the safe reading.
        'config.providerSubject': providerSubject || null,
        revokedAt: null,
        errorMessage: null,
      },
    },
    { new: true },
  );
  if (!published) return callbackRedirect(res, 'error', 'state_consumed');

  // Also discard any assessment started during the exchange, before the row
  // update made the new grant visible.
  forgetHostedMcpDrift(String(consumed._id));

  return callbackRedirect(res, 'connected', undefined, manualRevokeAt ? {
    manualRevoke: {
      entryId: entry.id,
      revokeAt: manualRevokeAt,
    },
  } : undefined);
};

router.get('/:entryId/callback', callbackRateLimit, async (req: Request, res: Response) => {
  const flow: CallbackFlow = {};
  try {
    await finishCallback(req, res, flow);
  } catch {
    // A new pending row can be marked failed only if no newer start has put a
    // different pending authorization on the same row.
    if (flow.pendingRowId) {
      try {
        await Integration.findOneAndUpdate(
          // A newer /start may have written a fresh pendingAuth while this
          // callback was waiting for the secret store. Leave that flow intact.
          {
            _id: flow.pendingRowId,
            status: 'pending',
            'config.pendingAuth': { $exists: false },
            'config.authGeneration': flow.authGeneration === undefined ? { $exists: false } : flow.authGeneration,
          },
          {
            $set: {
              status: 'error',
              errorMessage: 'Connection could not be completed. Try again.',
              errorMessageUserFacing: true,
            },
            $unset: { 'config.pendingAuth': 1 },
          },
          { new: true },
        );
      } catch {
        // A failed recovery write still must not strand the browser or reject
        // the Express handler.
      }
    }
    if (flow.connectedRowId && flow.credentialWriteStarted) {
      try {
        // put replaces material behind a stable ref. If a later write fails,
        // the old access/refresh pair may be mixed; retire the connection so
        // no caller can use either token until a fresh authorization succeeds.
        await Integration.findOneAndUpdate(
          {
            _id: flow.connectedRowId,
            status: 'connected',
            'config.authGeneration': flow.authGeneration === undefined ? { $exists: false } : flow.authGeneration,
          },
          {
            $set: {
              status: 'error',
              errorMessage: 'Connection could not be completed. Try again.',
              errorMessageUserFacing: true,
            },
          },
        );
      } catch {
        // Keep the HTTP callback contained even if recovery storage fails.
      }
    }
    console.error('[hosted-mcp] callback failed', JSON.stringify({ entryId: req.params.entryId }));
    if (!res.headersSent) callbackRedirect(res, 'error', 'connector_unavailable');
  }
});

module.exports = router;