Phase 2 — Google Ads Adapter

A. Cache layer ──────────────┐
                             │
B. Envelope encryption ──┬───┼──▶ D. Google Ads adapter ──▶ E. Seven tools
                         │   │           │
                         ▼   │           ├──▶ F. Token revocation endpoint
                  C. OAuth flow          │
                  (uses Phase 1's        ├──▶ G. DEK rotation tooling (depends on B)
                   oauth-state)          │
                                         └──▶ H. Cache metrics (depends on A)

I. Tests run alongside everything

export const TTL_SECONDS = {
  google: {
    account_health:     3600,    // 1h
    search_term_waste:  7200,    // 2h
    quality_score:      7200,
    auction_insights:   14_400,  // 4h
    pmax_breakdown:     7200,
    budget_optimizer:   3600,
    weekly_anomaly:     7200,
  },
  meta:   { account_health: 3600 }, // populated in Phase 3
  tiktok: { account_health: 7200 },
} as const;
 
export type Platform   = keyof typeof TTL_SECONDS;
export type ReportType = keyof (typeof TTL_SECONDS)['google'];
 
export function ttlFor(platform: Platform, reportType: ReportType): number {
  const ttl = (TTL_SECONDS[platform] as Record<string, number>)[reportType];
  if (!ttl) throw new Error(`No TTL configured for ${platform}/${reportType}`);
  return ttl;
}

import { and, eq, gt, sql } from 'drizzle-orm';
import { db } from '../db/index.js';
import { metricCache } from '../db/schema.js';
import { ttlFor, type Platform, type ReportType } from './ttl-config.js';
import { recordCacheEvent } from './metrics.js';
 
export interface CacheKey {
  tenantId: string;
  platform: Platform;
  reportType: ReportType;
  dateRangeKey: string;
}
 
export async function readOrFetch<T>(
  key: CacheKey,
  fetcher: () => Promise<T>,
  opts: { force?: boolean } = {},
): Promise<{ data: T; cache: 'hit' | 'miss' }> {
  const now = new Date();
  if (!opts.force) {
    const [hit] = await db.select().from(metricCache).where(and(
      eq(metricCache.tenantId,     key.tenantId),
      eq(metricCache.platform,     key.platform),
      eq(metricCache.reportType,   key.reportType),
      eq(metricCache.dateRangeKey, key.dateRangeKey),
      gt(metricCache.expiresAt,    now),
    ));
    if (hit) {
      recordCacheEvent('hit', key);
      return { data: hit.data as T, cache: 'hit' };
    }
  }
 
  // pg_advisory_xact_lock prevents the thundering-herd effect: if 50 concurrent
  // requests miss the same key, only one fetches; the others wait, then read the
  // freshly-written row on retry.
  return await db.transaction(async (tx) => {
    const lockKey = hashCacheKey(key); // bigint
    await tx.execute(sql`SELECT pg_advisory_xact_lock(${lockKey})`);
 
    // Re-check after acquiring lock (double-checked locking) — skipped on force,
    // since the eager-sync worker explicitly wants to overwrite a still-fresh row.
    if (!opts.force) {
      const [retry] = await tx.select().from(metricCache).where(and(
        eq(metricCache.tenantId,     key.tenantId),
        eq(metricCache.platform,     key.platform),
        eq(metricCache.reportType,   key.reportType),
        eq(metricCache.dateRangeKey, key.dateRangeKey),
        gt(metricCache.expiresAt,    new Date()),
      ));
      if (retry) {
        recordCacheEvent('hit', key);
        return { data: retry.data as T, cache: 'hit' };
      }
    }
 
    recordCacheEvent('miss', key);
    const fresh = await fetcher();
    const expiresAt = new Date(Date.now() + ttlFor(key.platform, key.reportType) * 1000);
    await tx.insert(metricCache).values({
      tenantId: key.tenantId, platform: key.platform,
      reportType: key.reportType, dateRangeKey: key.dateRangeKey,
      data: fresh as object, expiresAt,
    }).onConflictDoUpdate({
      target: [metricCache.tenantId, metricCache.platform, metricCache.reportType, metricCache.dateRangeKey],
      set: { data: fresh as object, expiresAt, fetchedAt: new Date() },
    });
    return { data: fresh, cache: 'miss' as const };
  });
}
 
function hashCacheKey(k: CacheKey): bigint {
  // Stable 64-bit hash for advisory lock — any deterministic hash works.
  const s = `${k.tenantId}|${k.platform}|${k.reportType}|${k.dateRangeKey}`;
  let h = 0xcbf29ce484222325n; const prime = 0x100000001b3n;
  for (const ch of Buffer.from(s)) { h ^= BigInt(ch); h = (h * prime) & 0xffffffffffffffffn; }
  // pg_advisory_lock takes int8 → cast to signed range
  return h > 0x7fffffffffffffffn ? h - 0x10000000000000000n : h;
}

import { createCipheriv, createDecipheriv, randomBytes } from 'node:crypto';
import { eq } from 'drizzle-orm';
import { db } from '../db/index.js';
import { tenantDeks } from '../db/schema.js';      // new table — see B3
import { loadSecret } from './secrets.loader.js';
 
const KEK = await loadSecret('CREDENTIAL_KEK');     // 32 bytes
if (KEK.length !== 32) throw new Error('CREDENTIAL_KEK must be exactly 32 bytes');
 
interface DekRow { tenantId: string; dekEnc: Buffer; dekIv: Buffer; dekTag: Buffer; kekVersion: number }
 
const dekCache = new Map<string, Buffer>(); // tenantId → plaintext DEK (process-local only)
 
async function getOrCreateDek(tenantId: string): Promise<Buffer> {
  const cached = dekCache.get(tenantId);
  if (cached) return cached;
 
  const [row] = await db.select().from(tenantDeks).where(eq(tenantDeks.tenantId, tenantId));
  if (row) {
    const dek = decryptWithKek(row.dekEnc, row.dekIv, row.dekTag);
    dekCache.set(tenantId, dek);
    return dek;
  }
 
  // First use for this tenant — generate a fresh DEK and wrap it under the KEK.
  const dek = randomBytes(32);
  const { ciphertext, iv, tag } = encryptWithKek(dek);
  await db.insert(tenantDeks).values({
    tenantId, dekEnc: ciphertext, dekIv: iv, dekTag: tag, kekVersion: 1,
  });
  dekCache.set(tenantId, dek);
  return dek;
}
 
function encryptWithKek(plaintext: Buffer) {
  const iv = randomBytes(12);
  const cipher = createCipheriv('aes-256-gcm', KEK, iv);
  const ciphertext = Buffer.concat([cipher.update(plaintext), cipher.final()]);
  return { ciphertext, iv, tag: cipher.getAuthTag() };
}
 
function decryptWithKek(ciphertext: Buffer, iv: Buffer, tag: Buffer): Buffer {
  const decipher = createDecipheriv('aes-256-gcm', KEK, iv);
  decipher.setAuthTag(tag);
  return Buffer.concat([decipher.update(ciphertext), decipher.final()]);
}
 
// Format on disk: iv(12) | tag(16) | ciphertext, base64
export async function encryptToken(tenantId: string, plaintext: string): Promise<string> {
  const dek = await getOrCreateDek(tenantId);
  const iv  = randomBytes(12);
  const cipher = createCipheriv('aes-256-gcm', dek, iv);
  const ct = Buffer.concat([cipher.update(plaintext, 'utf8'), cipher.final()]);
  return Buffer.concat([iv, cipher.getAuthTag(), ct]).toString('base64');
}
 
export async function decryptToken(tenantId: string, blob: string): Promise<string> {
  const dek = await getOrCreateDek(tenantId);
  const buf = Buffer.from(blob, 'base64');
  const iv  = buf.subarray(0, 12);
  const tag = buf.subarray(12, 28);
  const ct  = buf.subarray(28);
  const decipher = createDecipheriv('aes-256-gcm', dek, iv);
  decipher.setAuthTag(tag);
  return Buffer.concat([decipher.update(ct), decipher.final()]).toString('utf8');
}
 
export async function destroyDek(tenantId: string): Promise<void> {
  await db.delete(tenantDeks).where(eq(tenantDeks.tenantId, tenantId));
  dekCache.delete(tenantId);
}

export const tenantDeks = pgTable('tenant_deks', {
  tenantId:   uuid('tenant_id').primaryKey().references(() => tenants.id, { onDelete: 'cascade' }),
  dekEnc:     bytea('dek_enc').notNull(),
  dekIv:      bytea('dek_iv').notNull(),
  dekTag:     bytea('dek_tag').notNull(),
  kekVersion: integer('kek_version').notNull().default(1),  // for B-series KEK rotation
  createdAt:  timestamp('created_at').defaultNow().notNull(),
  rotatedAt:  timestamp('rotated_at'),
});

ALTER TABLE tenant_deks ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation_deks ON tenant_deks
  USING (tenant_id = current_setting('app.current_tenant_id', true)::uuid);

export const config = z.object({
  GOOGLE_CLIENT_ID:           z.string().min(1),
  GOOGLE_OAUTH_REDIRECT_URI:  z.string().url(),
  GOOGLE_AUTH_ENDPOINT:       z.string().url().default('https://accounts.google.com/o/oauth2/v2/auth'),
  GOOGLE_TOKEN_ENDPOINT:      z.string().url().default('https://oauth2.googleapis.com/token'),
  GOOGLE_SCOPES:              z.string().default('https://www.googleapis.com/auth/adwords'),
}).parse(process.env);

import type { FastifyPluginAsync } from 'fastify';
import { z } from 'zod';
import { createOAuthState, consumeOAuthState } from './oauth-state.service.js';
import { config } from '../config.js';
import { writeAuditEvent } from '../security/audit-log.service.js';
import { getAdapter } from '../adapters/registry.js';
 
const StartParams = z.object({ platform: z.enum(['google', 'meta', 'tiktok']) });
 
export const oauthRoutes: FastifyPluginAsync = async (fastify) => {
  fastify.get('/auth/:platform/start', async (req, reply) => {
    const { platform } = StartParams.parse(req.params);
    if (platform !== 'google') return reply.code(501).send({ error: 'unsupported_platform' }); // Phase 3 lifts this
    if (!req.tenantId) return reply.code(401).send();
 
    const { state, codeChallenge } = await createOAuthState(req.tenantId, platform);
    await writeAuditEvent('oauth.flow_started', 'success', { tenantId: req.tenantId, platform });
 
    const url = new URL(config.GOOGLE_AUTH_ENDPOINT);
    url.searchParams.set('client_id',             config.GOOGLE_CLIENT_ID);
    url.searchParams.set('redirect_uri',          config.GOOGLE_OAUTH_REDIRECT_URI);
    url.searchParams.set('response_type',         'code');
    url.searchParams.set('scope',                 config.GOOGLE_SCOPES);
    url.searchParams.set('access_type',           'offline');                 // get refresh_token
    url.searchParams.set('prompt',                'consent');                 // force refresh_token issuance
    url.searchParams.set('state',                 state);
    url.searchParams.set('code_challenge',        codeChallenge);
    url.searchParams.set('code_challenge_method', 'S256');
 
    return reply.redirect(url.toString(), 302);
  });
 
  fastify.get('/auth/:platform/callback', async (req, reply) => {
    const { platform } = StartParams.parse(req.params);
    const Q = z.object({ code: z.string().min(1), state: z.string().min(1), error: z.string().optional() });
    const q = Q.parse(req.query);
    if (q.error) {
      await writeAuditEvent('oauth.flow_failed', 'failure', { platform, reason: q.error });
      return reply.code(400).send({ error: 'oauth_denied' });
    }
 
    const stateRow = await consumeOAuthState(q.state); // single-use, throws on miss/expiry
    if (stateRow.platform !== platform) {
      await writeAuditEvent('oauth.flow_failed', 'failure', { reason: 'state_platform_mismatch' });
      return reply.code(400).send({ error: 'invalid_state' });
    }
 
    const adapter = getAdapter(platform);
    await adapter.exchangeCode(stateRow.tenantId, q.code, stateRow.codeVerifier);
    await writeAuditEvent('oauth.flow_completed', 'success', { tenantId: stateRow.tenantId, platform });
 
    return reply.send({ status: 'connected', platform });
  });
};

export interface PlatformAdapter {
  readonly platform: 'google' | 'meta' | 'tiktok';
  exchangeCode(tenantId: string, code: string, codeVerifier: string): Promise<void>;
  ensureValidToken(tenantId: string): Promise<string>;
  fetchAccountHealth(tenantId: string, range: DateRange): Promise<AccountHealthData>;
  fetchCampaigns(tenantId: string, range: DateRange): Promise<CampaignData[]>;
  fetchSearchTerms(tenantId: string, range: DateRange): Promise<SearchTermData[]>;
  fetchAuctionInsights(tenantId: string, range: DateRange): Promise<AuctionData>;
  fetchKeywordQualityScores(tenantId: string, range: DateRange): Promise<QSData[]>;
  fetchAssetGroups(tenantId: string, range: DateRange): Promise<AssetGroupData[]>;
  revoke(tenantId: string): Promise<void>;     // pulled in for §F
}
 
export type DateRange = 'last_7_days' | 'last_30_days' | 'last_90_days';

import { GoogleAdapter } from './google/index.js';
import type { PlatformAdapter } from './adapter.interface.js';
 
const adapters = new Map<string, PlatformAdapter>([
  ['google', new GoogleAdapter()],
  // Phase 3: Meta, TikTok
]);
 
export function getAdapter(platform: string): PlatformAdapter {
  const a = adapters.get(platform);
  if (!a) throw new Error(`unsupported_platform: ${platform}`);
  return a;
}

import { eq } from 'drizzle-orm';
import { db } from '../../db/index.js';
import { platformCredentials } from '../../db/schema.js';
import { encryptToken, decryptToken } from '../../security/credentials.service.js';
import { loadSecret } from '../../security/secrets.loader.js';
import { config } from '../../config.js';
import { writeAuditEvent } from '../../security/audit-log.service.js';
 
const REFRESH_SKEW_SECONDS = 300; // refresh 5 min before actual expiry
 
export async function exchangeAuthCode(tenantId: string, code: string, codeVerifier: string): Promise<void> {
  const clientSecret = (await loadSecret('GOOGLE_CLIENT_SECRET' as never)).toString('utf8');
  const body = new URLSearchParams({
    code,
    client_id:     config.GOOGLE_CLIENT_ID,
    client_secret: clientSecret,
    redirect_uri:  config.GOOGLE_OAUTH_REDIRECT_URI,
    grant_type:    'authorization_code',
    code_verifier: codeVerifier,
  });
  const res = await fetch(config.GOOGLE_TOKEN_ENDPOINT, {
    method: 'POST',
    headers: { 'content-type': 'application/x-www-form-urlencoded' },
    body,
  });
  if (!res.ok) throw new Error(`google_token_exchange_failed: ${res.status}`);
  const tokens = (await res.json()) as { access_token: string; refresh_token: string; expires_in: number; scope: string };
 
  const accessEnc  = await encryptToken(tenantId, tokens.access_token);
  const refreshEnc = await encryptToken(tenantId, tokens.refresh_token);
 
  // Customer ID resolution — required for Google Ads API. Phase 2 simplification:
  // store empty string, populate on first fetch via listAccessibleCustomers.
  await db.insert(platformCredentials).values({
    tenantId, platform: 'google', accountId: '',
    accessTokenEnc: accessEnc, refreshTokenEnc: refreshEnc,
    tokenExpiresAt: new Date(Date.now() + tokens.expires_in * 1000),
    scopes: tokens.scope.split(' '),
  }).onConflictDoUpdate({
    target: [platformCredentials.tenantId, platformCredentials.platform],
    set: {
      accessTokenEnc: accessEnc, refreshTokenEnc: refreshEnc,
      tokenExpiresAt: new Date(Date.now() + tokens.expires_in * 1000),
      scopes: tokens.scope.split(' '),
      updatedAt: new Date(),
    },
  });
}
 
export async function ensureValidToken(tenantId: string): Promise<string> {
  const [row] = await db.select().from(platformCredentials)
    .where(eq(platformCredentials.tenantId, tenantId));
  if (!row || row.platform !== 'google') throw new Error('no_google_credentials');
 
  const expiresAt = row.tokenExpiresAt!.getTime();
  if (expiresAt - Date.now() > REFRESH_SKEW_SECONDS * 1000) {
    return decryptToken(tenantId, row.accessTokenEnc);
  }
 
  // Lazy refresh
  const refresh = await decryptToken(tenantId, row.refreshTokenEnc!);
  const clientSecret = (await loadSecret('GOOGLE_CLIENT_SECRET' as never)).toString('utf8');
  const body = new URLSearchParams({
    refresh_token: refresh,
    client_id:     config.GOOGLE_CLIENT_ID,
    client_secret: clientSecret,
    grant_type:    'refresh_token',
  });
  const res = await fetch(config.GOOGLE_TOKEN_ENDPOINT, {
    method: 'POST', headers: { 'content-type': 'application/x-www-form-urlencoded' }, body,
  });
  if (!res.ok) {
    await writeAuditEvent('oauth.token_refreshed', 'failure', { tenantId, platform: 'google', status: res.status });
    throw new Error(`google_token_refresh_failed: ${res.status}`);
  }
  const tokens = (await res.json()) as { access_token: string; expires_in: number };
  const newAccessEnc = await encryptToken(tenantId, tokens.access_token);
 
  await db.update(platformCredentials)
    .set({ accessTokenEnc: newAccessEnc, tokenExpiresAt: new Date(Date.now() + tokens.expires_in * 1000), updatedAt: new Date() })
    .where(eq(platformCredentials.tenantId, tenantId));
  await writeAuditEvent('oauth.token_refreshed', 'success', { tenantId, platform: 'google' });
 
  return tokens.access_token;
}

import { GoogleAdsApi } from 'google-ads-api';
import { loadSecret } from '../../security/secrets.loader.js';
import { config } from '../../config.js';
import { ensureValidToken } from './auth.js';
import type { DateRange } from '../adapter.interface.js';
 
const developerToken = (await loadSecret('GOOGLE_DEVELOPER_TOKEN' as never)).toString('utf8');
const clientSecret   = (await loadSecret('GOOGLE_CLIENT_SECRET' as never)).toString('utf8');
 
const apiClient = new GoogleAdsApi({
  client_id: config.GOOGLE_CLIENT_ID,
  client_secret: clientSecret,
  developer_token: developerToken,
});
 
const RANGE_GAQL: Record<DateRange, string> = {
  last_7_days:  'segments.date DURING LAST_7_DAYS',
  last_30_days: 'segments.date DURING LAST_30_DAYS',
  last_90_days: 'segments.date DURING LAST_90_DAYS',
};
 
async function customer(tenantId: string, customerId: string) {
  return apiClient.Customer({
    customer_id: customerId,
    refresh_token: '__handled_via_ensureValidToken__', // adapter-level shim — see D4
  });
}
 
export async function queryAccountHealth(tenantId: string, customerId: string, range: DateRange) {
  await ensureValidToken(tenantId); // refreshes if needed; library uses access_token via shim
  const c = await customer(tenantId, customerId);
  const rows = await c.report({
    entity: 'campaign',
    attributes: ['campaign.id', 'campaign.name', 'campaign.status'],
    metrics:    ['metrics.cost_micros', 'metrics.conversions_value', 'metrics.conversions',
                 'metrics.clicks',      'metrics.impressions'],
    constraints: { 'campaign.status': 'ENABLED' },
    date_constant: range === 'last_7_days' ? 'LAST_7_DAYS'
                  : range === 'last_30_days' ? 'LAST_30_DAYS'
                  : 'LAST_90_DAYS',
  });
  return rows; // shape captured in src/adapters/google/types.ts (Zod-validated)
}
 
// queryPmaxBreakdown, querySearchTerms, queryAuctionInsights, queryQualityScore,
// queryBudgetSplit, queryWeeklyAnomaly — same pattern, GAQL string varies per report.

const accessToken = await ensureValidToken(tenantId);
return apiClient.Customer({
  customer_id: customerId,
  login_customer_id: customerId, // for MCC; same as customer_id for direct accounts
  oauth_access_token: accessToken,
});

export async function resolveCustomerId(tenantId: string): Promise<string> {
  const [row] = await db.select().from(platformCredentials)
    .where(and(eq(platformCredentials.tenantId, tenantId), eq(platformCredentials.platform, 'google')));
  if (row.accountId) return row.accountId;
 
  const accessToken = await ensureValidToken(tenantId);
  const res = await fetch('https://googleads.googleapis.com/v17/customers:listAccessibleCustomers', {
    headers: {
      'Authorization': `Bearer ${accessToken}`,
      'developer-token': developerToken,
    },
  });
  const json = (await res.json()) as { resourceNames: string[] };
  const first = json.resourceNames[0]?.split('/')[1];
  if (!first) throw new Error('no_accessible_customers');
 
  await db.update(platformCredentials).set({ accountId: first })
    .where(eq(platformCredentials.tenantId, tenantId));
  return first;
}

import type { PlatformAdapter, DateRange } from '../adapter.interface.js';
import { exchangeAuthCode, ensureValidToken } from './auth.js';
import { resolveCustomerId } from './customers.js';
import { revokeUpstream } from './revoke.js';            // §F
import * as q from './queries.js';
 
export class GoogleAdapter implements PlatformAdapter {
  readonly platform = 'google' as const;
 
  exchangeCode = exchangeAuthCode;
  ensureValidToken = ensureValidToken;
 
  async fetchAccountHealth(tenantId: string, range: DateRange) {
    const customerId = await resolveCustomerId(tenantId);
    return q.queryAccountHealth(tenantId, customerId, range);
  }
  // fetchCampaigns / fetchSearchTerms / etc — same shape
  async revoke(tenantId: string) { await revokeUpstream(tenantId); }
}

import type { z } from 'zod';
import { readOrFetch } from '../../cache/cache.service.js';
import { getAdapter } from '../../adapters/registry.js';
import { writeAuditEvent } from '../../security/audit-log.service.js';
import type { Platform, ReportType } from '../../cache/ttl-config.js';
 
export function makeTool<I extends { platform: Platform; dateRange: 'last_7_days' | 'last_30_days' | 'last_90_days' }, O>(opts: {
  name: string;
  description: string;
  inputSchema: z.ZodType<I>;
  reportType: ReportType;
  supportedPlatforms: ReadonlyArray<Platform>;
  fetcher: (adapter: ReturnType<typeof getAdapter>, input: I) => Promise<O>;
}) {
  return {
    name: opts.name,
    description: opts.description,
    inputSchema: opts.inputSchema,
    async handler(input: I, ctx: { tenantId: string; requestId: string }) {
      if (!opts.supportedPlatforms.includes(input.platform)) {
        await writeAuditEvent('mcp.tool_failed', 'failure', { tenantId: ctx.tenantId, requestId: ctx.requestId, tool: opts.name, reason: 'unsupported_platform' });
        return { error: 'unsupported_platform', platform: input.platform } as const;
      }
      const adapter = getAdapter(input.platform);
      const { data, cache } = await readOrFetch({
        tenantId: ctx.tenantId,
        platform: input.platform,
        reportType: opts.reportType,
        dateRangeKey: input.dateRange,
      }, () => opts.fetcher(adapter, input));
      await writeAuditEvent('mcp.tool_called', 'success', { tenantId: ctx.tenantId, requestId: ctx.requestId, tool: opts.name, cache });
      return { data, cache };
    },
  };
}

// src/mcp/tools/account-health.ts
import { z } from 'zod';
import { makeTool } from './_helpers.js';
 
const Input = z.object({
  platform:  z.enum(['google', 'meta', 'tiktok']),
  dateRange: z.enum(['last_7_days', 'last_30_days', 'last_90_days']),
});
 
export const accountHealthTool = makeTool({
  name: 'get_account_health',
  description: 'Spend, ROAS, CPA, CTR — 90-day trends and per-campaign ranking.',
  inputSchema: Input,
  reportType: 'account_health',
  supportedPlatforms: ['google'], // Phase 3 adds meta, tiktok
  fetcher: (adapter, input) => adapter.fetchAccountHealth(input.dateRange === 'last_90_days' ? 'last_90_days' : input.dateRange),
});

import { and, eq } from 'drizzle-orm';
import { db } from '../../db/index.js';
import { platformCredentials } from '../../db/schema.js';
import { decryptToken } from '../../security/credentials.service.js';
 
export async function revokeUpstream(tenantId: string): Promise<void> {
  const [row] = await db.select().from(platformCredentials)
    .where(and(eq(platformCredentials.tenantId, tenantId), eq(platformCredentials.platform, 'google')));
  if (!row) return;
 
  const refresh = await decryptToken(tenantId, row.refreshTokenEnc!);
  // Revocation is best-effort — we still wipe the row even if Google is unreachable.
  try {
    await fetch(`https://oauth2.googleapis.com/revoke?token=${encodeURIComponent(refresh)}`, { method: 'POST' });
  } catch { /* swallow — wipe still happens */ }
 
  await db.delete(platformCredentials)
    .where(and(eq(platformCredentials.tenantId, tenantId), eq(platformCredentials.platform, 'google')));
}

fastify.post('/admin/tenants/:tenantId/disconnect/:platform', async (req, reply) => {
  if (req.headers['x-admin-token'] !== adminToken) return reply.code(401).send();
  const P = z.object({ tenantId: z.string().uuid(), platform: z.enum(['google', 'meta', 'tiktok']) });
  const p = P.parse(req.params);
 
  const adapter = getAdapter(p.platform);
  await adapter.revoke(p.tenantId);
  await writeAuditEvent('oauth.token_revoked', 'success', { tenantId: p.tenantId, platform: p.platform });
  return { status: 'disconnected' };
});

import { eq } from 'drizzle-orm';
import { randomBytes } from 'node:crypto';
import { db } from '../src/db/index.js';
import { tenantDeks, platformCredentials } from '../src/db/schema.js';
import { decryptToken, encryptToken } from '../src/security/credentials.service.js';
 
const tenantId = process.argv[2];
if (!tenantId) throw new Error('Usage: tsx scripts/rotate-dek.ts <tenantId>');
 
await db.transaction(async (tx) => {
  // 1. Read all encrypted artifacts under the OLD DEK.
  const [creds] = await tx.select().from(platformCredentials).where(eq(platformCredentials.tenantId, tenantId));
  const accessPlain  = creds ? await decryptToken(tenantId, creds.accessTokenEnc) : null;
  const refreshPlain = creds?.refreshTokenEnc ? await decryptToken(tenantId, creds.refreshTokenEnc) : null;
 
  // 2. Destroy the cached + persisted DEK so the next encrypt() call generates a new one.
  await tx.delete(tenantDeks).where(eq(tenantDeks.tenantId, tenantId));
  // 3. Clear the in-memory cache.
  (await import('../src/security/credentials.service.js') as any).dekCache.delete?.(tenantId);
 
  // 4. Re-encrypt.
  if (creds && accessPlain) {
    const accessEnc  = await encryptToken(tenantId, accessPlain);
    const refreshEnc = refreshPlain ? await encryptToken(tenantId, refreshPlain) : creds.refreshTokenEnc;
    await tx.update(platformCredentials)
      .set({ accessTokenEnc: accessEnc, refreshTokenEnc: refreshEnc, updatedAt: new Date() })
      .where(eq(platformCredentials.tenantId, tenantId));
  }
  await tx.update(tenantDeks).set({ rotatedAt: new Date() }).where(eq(tenantDeks.tenantId, tenantId));
});
 
console.log(`Rotated DEK for tenant ${tenantId}`);

import type { CacheKey } from './cache.service.js';
 
interface Counter { hit: number; miss: number }
const counters = new Map<string, Counter>(); // key: `${platform}/${reportType}`
 
export function recordCacheEvent(kind: 'hit' | 'miss', key: Pick<CacheKey, 'platform' | 'reportType'>): void {
  const k = `${key.platform}/${key.reportType}`;
  const c = counters.get(k) ?? { hit: 0, miss: 0 };
  c[kind] += 1;
  counters.set(k, c);
}
 
export function snapshotCacheMetrics(): Record<string, Counter & { hitRate: number }> {
  const out: Record<string, Counter & { hitRate: number }> = {};
  for (const [k, c] of counters) {
    const total = c.hit + c.miss;
    out[k] = { ...c, hitRate: total === 0 ? 0 : c.hit / total };
  }
  return out;
}

fastify.get('/admin/metrics/cache', async (req, reply) => {
  if (req.headers['x-admin-token'] !== adminToken) return reply.code(401).send();
  return snapshotCacheMetrics();
});

# 0. Phase 1 already up: docker compose up -d postgres, secrets seeded, server running on 127.0.0.1:3001.
 
# 1. Apply Phase 2 migration (adds tenant_deks + cache unique index)
npm run db:migrate
 
# 2. Seed a tenant + API key, capture both
node scripts/seed-tenant.mjs
export KEY=...; export TENANT_ID=...
 
# 3. Start the OAuth flow — should redirect (302) to Google
curl -i -H "X-Api-Key: $KEY" http://127.0.0.1:3001/auth/google/start
# → HTTP/1.1 302 ... Location: https://accounts.google.com/o/oauth2/v2/auth?...code_challenge=...
 
# 4. Complete consent in a browser; the callback writes to platform_credentials.
# Verify the row is encrypted (no plaintext "ya29...." substring):
psql "postgresql://mcp_admin:dev_only_password@127.0.0.1:5432/deneva_mcp" \
  -c "SELECT length(access_token_enc), substring(access_token_enc, 1, 20) FROM platform_credentials;"
 
# 5. Call get_account_health — first call should be a cache MISS and a real Google fetch
curl -X POST http://127.0.0.1:3001/mcp \
  -H "X-Api-Key: $KEY" -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_account_health","arguments":{"platform":"google","dateRange":"last_7_days"}}}'
# → response includes "cache":"miss"
 
# 6. Same call again — cache HIT
curl -X POST http://127.0.0.1:3001/mcp -H "X-Api-Key: $KEY" -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_account_health","arguments":{"platform":"google","dateRange":"last_7_days"}}}'
# → response includes "cache":"hit"
 
# 7. Try the same tool with platform=meta — clean unsupported_platform error
curl -X POST http://127.0.0.1:3001/mcp -H "X-Api-Key: $KEY" -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_account_health","arguments":{"platform":"meta","dateRange":"last_7_days"}}}'
# → {"error":"unsupported_platform","platform":"meta"}
 
# 8. Inspect cache metrics
curl -H "x-admin-token: $ADMIN_TOKEN" http://127.0.0.1:3001/admin/metrics/cache
# → {"google/account_health":{"hit":1,"miss":1,"hitRate":0.5}}
 
# 9. Rotate the DEK
npx tsx scripts/rotate-dek.ts $TENANT_ID
# → "Rotated DEK for tenant ..."
 
# 10. Re-run step 6 — the cached row was encrypted with the OLD DEK, but cache row
# stores plaintext metric data, not tokens. Tokens are decrypted via the NEW DEK
# without issue. Expect the same response.
 
# 11. Disconnect the tenant
curl -X POST -H "x-admin-token: $ADMIN_TOKEN" http://127.0.0.1:3001/admin/tenants/$TENANT_ID/disconnect/google
# → {"status":"disconnected"}
psql "..." -c "SELECT count(*) FROM platform_credentials WHERE tenant_id = '$TENANT_ID';"
# → 0
 
# 12. Inspect the audit trail
psql "..." -c "SELECT event_type, outcome, count(*) FROM audit_log WHERE tenant_id = '$TENANT_ID' GROUP BY 1,2 ORDER BY 1,2;"
# → rows for: oauth.flow_started, oauth.flow_completed, oauth.token_refreshed (if expiry crossed),
#            mcp.tool_called (×2), api_key.rotated (if you rotated), oauth.token_revoked