diff --git a/.changeset/registry-crawler-cache.md b/.changeset/registry-crawler-cache.md new file mode 100644 index 0000000000..544c954038 --- /dev/null +++ b/.changeset/registry-crawler-cache.md @@ -0,0 +1,16 @@ +--- +--- + +Crawler now caches successful adagents.json fetches into the publishers +overlay table (migration 432) and projects parsed properties into +catalog_properties + catalog_identifiers in the same transaction. The +existing discovered_properties / agent_property_authorizations writes +continue alongside the new ones for one release as a fallback before PR 5 +of #3177 drops the old tables. + +Closes the gap surfaced by Setupad escalation #218: properties that landed +in discovered_properties via the crawler never made it into the catalog +because migration 336 was a one-time seed. Every successful crawl now +lands in both places. + +Refs #3177. Builds on #3195. diff --git a/server/src/crawler.ts b/server/src/crawler.ts index 427b57c219..b861ff7319 100644 --- a/server/src/crawler.ts +++ b/server/src/crawler.ts @@ -4,6 +4,7 @@ import { FederatedIndexService } from "./federated-index.js"; import { AdAgentsManager } from "./adagents-manager.js"; import { BrandManager } from "./brand-manager.js"; import { BrandDatabase } from "./db/brand-db.js"; +import { PublisherDatabase, type AdagentsManifest } from "./db/publisher-db.js"; import { MemberDatabase } from "./db/member-db.js"; import { CapabilityDiscovery } from "./capabilities.js"; import { HealthChecker } from "./health.js"; @@ -27,6 +28,7 @@ export class CrawlerService { private adAgentsManager: AdAgentsManager; private brandManager: BrandManager; private brandDb: BrandDatabase; + private publisherDb: PublisherDatabase; private memberDb: MemberDatabase; private capabilityDiscovery: CapabilityDiscovery; private healthChecker: HealthChecker; @@ -40,6 +42,7 @@ export class CrawlerService { this.adAgentsManager = new AdAgentsManager(); this.brandManager = new BrandManager(); this.brandDb = new BrandDatabase(); + this.publisherDb = new PublisherDatabase(); this.memberDb = new MemberDatabase(); this.capabilityDiscovery = new CapabilityDiscovery(); this.healthChecker = new HealthChecker(); @@ -378,6 +381,8 @@ export class CrawlerService { const propCount = validation.raw_data.properties?.length || 0; log.debug({ domain: pubConfig.domain, agentCount, propCount }, 'Domain crawled'); + await this.cacheAdagentsManifest(pubConfig.domain, validation.raw_data as AdagentsManifest); + // Record agents for (const authorizedAgent of validation.raw_data.authorized_agents) { if (!authorizedAgent.url) continue; @@ -430,6 +435,8 @@ export class CrawlerService { await this.federatedIndex.markPublisherHasValidAdagents(domain); processedDomains.add(domain); + await this.cacheAdagentsManifest(domain, validation.raw_data as AdagentsManifest); + for (const authorizedAgent of validation.raw_data.authorized_agents) { if (!authorizedAgent.url) continue; @@ -603,6 +610,25 @@ export class CrawlerService { return this.federatedIndex; } + /** + * Cache a validated adagents.json manifest into the publishers overlay and + * project its properties into the property catalog. The writer runs as one + * transaction with per-property savepoints, so a malformed property is + * skipped without losing the rest of the manifest. + * + * Failure of the entire transaction is logged but does not abort the crawl: + * the legacy discovered_properties / agent_property_authorizations writes + * happen separately via recordPropertiesForAgent and remain authoritative + * until catalog readers take over. + */ + private async cacheAdagentsManifest(domain: string, manifest: AdagentsManifest): Promise { + try { + await this.publisherDb.upsertAdagentsCache({ domain, manifest }); + } catch (err) { + log.warn({ domain, err: err instanceof Error ? err.message : err }, 'Publisher cache write failed'); + } + } + /** * Record properties from adagents.json and link them to an agent. * If the agent has property_ids specified, only record those specific properties. @@ -827,6 +853,8 @@ export class CrawlerService { return; } + await this.cacheAdagentsManifest(domain, validation.raw_data as AdagentsManifest); + // Record agents and properties for (const authorizedAgent of validation.raw_data.authorized_agents) { if (!authorizedAgent.url) continue; @@ -994,6 +1022,8 @@ export class CrawlerService { const validation = await this.adAgentsManager.validateDomain(domain); if (!validation.valid || !validation.raw_data?.authorized_agents) return; + await this.cacheAdagentsManifest(domain, validation.raw_data as AdagentsManifest); + for (const authorizedAgent of validation.raw_data.authorized_agents) { if (!authorizedAgent.url) continue; diff --git a/server/src/db/publisher-db.ts b/server/src/db/publisher-db.ts new file mode 100644 index 0000000000..644ad2e809 --- /dev/null +++ b/server/src/db/publisher-db.ts @@ -0,0 +1,305 @@ +import { getClient } from './client.js'; +import { uuidv7 } from './uuid.js'; +import { normalizeIdentifier } from '../services/identifier-normalization.js'; +import { createLogger } from '../logger.js'; +import type { PoolClient } from 'pg'; + +const log = createLogger('publisher-db'); + +/** + * Property as it appears inside an adagents.json file. The manifest body is + * untrusted (publisher-controlled), so callers should pass already-validated + * data — see AdAgentsManager.validateDomain. Fields here are loose because + * downstream projection only needs identifiers + property_id. + */ +export interface AdagentsProperty { + property_id?: string; + property_type?: string; + name?: string; + identifiers?: Array<{ type?: string; value?: string }>; + tags?: string[]; +} + +export interface AdagentsManifest { + authorized_agents?: unknown; + properties?: AdagentsProperty[]; + [key: string]: unknown; +} + +export interface UpsertAdagentsCacheInput { + domain: string; + manifest: AdagentsManifest; + expiresAt?: Date; +} + +const ADAGENTS_CREATED_BY_PREFIX = 'adagents_json:'; + +function adagentsCreatedBy(publisherDomain: string): string { + return `${ADAGENTS_CREATED_BY_PREFIX}${publisherDomain}`; +} + +/** + * Whether a domain/subdomain identifier lexically belongs to the publisher. + * + * Bundle IDs, RSS URLs, and other non-domain identifier types have no + * lexical relationship to the publisher's hostname, so they are never + * anchors. This is what stops a manifest hosted at attacker.example from + * legitimately claiming `domain:victim.example` — the anchor check rejects + * the cross-publisher domain claim before it can land in the catalog. + */ +function isPublisherDomainAnchor(publisherDomain: string, type: string, value: string): boolean { + if (type !== 'domain' && type !== 'subdomain') return false; + if (value === publisherDomain) return true; + return value.endsWith(`.${publisherDomain}`); +} + +/** + * Database operations for the publisher overlay (migration 432). + * + * Caches the source-of-truth adagents.json file body and projects the parsed + * manifest into the property catalog (catalog_properties + catalog_identifiers). + * The cache write and the per-property projections share one transaction; + * each property is wrapped in a savepoint so a constraint violation on one + * malformed property does not lose the rest of the manifest. + */ +export class PublisherDatabase { + /** + * Cache an adagents.json manifest and project its properties into the + * catalog. + * + * ON CONFLICT for the publishers row only touches the manifest body and the + * crawl-tracking columns; org/ownership and review state are preserved so a + * later org claim isn't wiped by a routine re-crawl. + */ + async upsertAdagentsCache(input: UpsertAdagentsCacheInput): Promise { + const domain = input.domain.toLowerCase(); + const client = await getClient(); + try { + await client.query('BEGIN'); + + await client.query( + `INSERT INTO publishers (domain, adagents_json, source_type, last_validated, expires_at) + VALUES ($1, $2::jsonb, 'adagents_json', NOW(), $3) + ON CONFLICT (domain) DO UPDATE SET + adagents_json = EXCLUDED.adagents_json, + source_type = 'adagents_json', + last_validated = NOW(), + expires_at = EXCLUDED.expires_at, + updated_at = NOW()`, + [domain, JSON.stringify(input.manifest), input.expiresAt ?? null] + ); + + const properties = Array.isArray(input.manifest.properties) ? input.manifest.properties : []; + for (let i = 0; i < properties.length; i += 1) { + const savepoint = `prop_${i}`; + await client.query(`SAVEPOINT ${savepoint}`); + try { + await this.projectPropertyToCatalog(client, domain, properties[i]); + await client.query(`RELEASE SAVEPOINT ${savepoint}`); + } catch (err) { + await client.query(`ROLLBACK TO SAVEPOINT ${savepoint}`); + log.warn( + { + domain, + propertyId: properties[i]?.property_id, + propertyIndex: i, + err: err instanceof Error ? err.message : err, + }, + 'Catalog projection failed for property; skipping' + ); + } + } + + await client.query('COMMIT'); + } catch (err) { + await client.query('ROLLBACK'); + throw err; + } finally { + client.release(); + } + } + + /** + * Project a single adagents.json property into catalog_properties + + * catalog_identifiers. Catalog rows are tagged + * evidence='adagents_json' / confidence='authoritative' so a property + * crawled now is indistinguishable from one seeded by migration 336. + * + * Tenant isolation rules — load-bearing for catalog correctness: + * + * 1. Cross-publisher domain claims are refused. A `domain` or `subdomain` + * identifier in the property must lexically belong to the publisher + * (equal to or a subdomain of publisherDomain). Otherwise the entire + * property is dropped — a manifest at attacker.example cannot land an + * authoritative claim for `domain:victim.example`. + * + * 2. Cross-publisher rid reuse is refused. If any matched rid was + * authored by another publisher's adagents.json, refuse — even + * ON CONFLICT DO NOTHING on the identifier insert wouldn't stop a + * reuse-branch UPDATE from rebinding adagents_url. + * + * 3. Multi-rid conflation is refused. If the identifier set spans + * multiple distinct existing rids, silent merging requires human + * review (catalog_disputes). + * + * 4. Foreign-rid reuse requires an anchor. If the matched rid was created + * by a non-adagents source (system seed, community, brand_json, member + * resolve), reuse is only allowed when the property carries at least + * one publisher-anchored identifier. Without this, an unanchored + * manifest can reach a seed rid via a bundle ID and overwrite its + * adagents_url via COALESCE. + * + * 5. Otherwise, reuse the single matching own-rid (re-crawl) or mint + * a new one. Identifiers go in with ON CONFLICT DO NOTHING so a + * non-anchor identifier already claimed by another rid silently + * drops rather than rebinding. + */ + private async projectPropertyToCatalog( + client: PoolClient, + publisherDomain: string, + property: AdagentsProperty, + ): Promise { + const rawIdentifiers = Array.isArray(property.identifiers) ? property.identifiers : []; + const identifiers = rawIdentifiers + .filter((i): i is { type: string; value: string } => + typeof i?.type === 'string' && typeof i?.value === 'string' && i.type.length > 0 && i.value.length > 0 + ) + .map((i) => { + const norm = normalizeIdentifier(i.type, i.value); + // catalog_identifiers.chk_identifier_lowercase requires the entire value + // to be lowercase. normalizeRssUrl preserves URL path case, and any + // future identifier type may also leak case; lowercase defensively to + // match the migration 336 seed and avoid silent rollbacks. + return { type: norm.type, value: norm.value.toLowerCase() }; + }); + + if (identifiers.length === 0) return; + + // Rule 1 — refuse cross-publisher domain claims. + const crossPublisherClaims = identifiers.filter( + (i) => + (i.type === 'domain' || i.type === 'subdomain') + && !isPublisherDomainAnchor(publisherDomain, i.type, i.value) + ); + if (crossPublisherClaims.length > 0) { + log.warn( + { + publisherDomain, + propertyId: property.property_id, + crossPublisherClaims, + }, + 'Catalog projection refused: property declares domain identifiers outside the publisher\'s domain' + ); + return; + } + + const hasAnchor = identifiers.some((i) => + isPublisherDomainAnchor(publisherDomain, i.type, i.value) + ); + + const tupleParams: unknown[] = []; + const tuplePlaceholders = identifiers + .map((ident, i) => { + tupleParams.push(ident.type, ident.value); + return `($${i * 2 + 1}, $${i * 2 + 2})`; + }) + .join(', '); + + // ORDER BY for determinism: when multiple distinct rids match, the same + // input always picks the same one (oldest first), so re-runs converge. + const existing = await client.query<{ property_rid: string; created_by: string | null }>( + `SELECT DISTINCT cp.property_rid, cp.created_by + FROM catalog_identifiers ci + JOIN catalog_properties cp ON cp.property_rid = ci.property_rid + WHERE (ci.identifier_type, ci.identifier_value) IN (${tuplePlaceholders}) + ORDER BY cp.created_by, cp.property_rid`, + tupleParams + ); + + const expectedCreatedBy = adagentsCreatedBy(publisherDomain); + + // Rule 2 — refuse cross-publisher rid reuse. + const conflicting = existing.rows.filter((r) => + typeof r.created_by === 'string' + && r.created_by.startsWith(ADAGENTS_CREATED_BY_PREFIX) + && r.created_by !== expectedCreatedBy + ); + if (conflicting.length > 0) { + log.warn( + { + publisherDomain, + propertyId: property.property_id, + conflictingCreatedBy: conflicting.map((r) => r.created_by), + conflictingRids: conflicting.map((r) => r.property_rid), + }, + 'Catalog projection refused: property identifiers are claimed by another publisher manifest' + ); + return; + } + + // Rule 3 — refuse multi-rid conflation. + const ownRids = Array.from(new Set(existing.rows.map((r) => r.property_rid))); + if (ownRids.length > 1) { + log.warn( + { publisherDomain, propertyId: property.property_id, rids: ownRids }, + 'Catalog projection refused: identifier set spans multiple existing properties (merge requires moderation)' + ); + return; + } + + const adagentsUrl = `https://${publisherDomain}/.well-known/adagents.json`; + let propertyRid: string; + + if (ownRids.length === 1) { + const matchedCreatedBy = existing.rows[0].created_by; + const isOwnRecrawl = matchedCreatedBy === expectedCreatedBy; + + // Rule 4 — foreign rid reuse requires an anchor. The publisher must + // produce a domain/subdomain identifier under their own domain to take + // ownership of (or update adagents_url on) a rid created by another + // source. Without this, a manifest declaring only a bundle ID could + // reach a seed rid via that bundle ID and rebind adagents_url. + if (!isOwnRecrawl && !hasAnchor) { + log.warn( + { + publisherDomain, + propertyId: property.property_id, + matchedCreatedBy, + matchedRid: ownRids[0], + }, + 'Catalog projection refused: cannot adopt a non-adagents rid without a publisher-anchored identifier' + ); + return; + } + + propertyRid = ownRids[0]; + await client.query( + `UPDATE catalog_properties SET + source_updated_at = NOW(), + updated_at = NOW(), + adagents_url = COALESCE(adagents_url, $2), + property_id = COALESCE(property_id, $3) + WHERE property_rid = $1`, + [propertyRid, adagentsUrl, property.property_id ?? null] + ); + } else { + propertyRid = uuidv7(); + await client.query( + `INSERT INTO catalog_properties + (property_rid, property_id, classification, source, status, adagents_url, created_by) + VALUES ($1, $2, 'property', 'authoritative', 'active', $3, $4)`, + [propertyRid, property.property_id ?? null, adagentsUrl, expectedCreatedBy] + ); + } + + for (const ident of identifiers) { + await client.query( + `INSERT INTO catalog_identifiers + (id, property_rid, identifier_type, identifier_value, evidence, confidence) + VALUES ($1, $2, $3, $4, 'adagents_json', 'authoritative') + ON CONFLICT (identifier_type, identifier_value) DO NOTHING`, + [uuidv7(), propertyRid, ident.type, ident.value] + ); + } + } +} diff --git a/server/tests/integration/registry-crawler-cache.test.ts b/server/tests/integration/registry-crawler-cache.test.ts new file mode 100644 index 0000000000..d8cfb5ad8d --- /dev/null +++ b/server/tests/integration/registry-crawler-cache.test.ts @@ -0,0 +1,738 @@ +/** + * Integration tests for PR 2 of #3177: the adagents.json crawler now caches + * the manifest into publishers (migration 432) and projects the parsed + * properties into catalog_properties + catalog_identifiers in the same + * transaction. + * + * The legacy discovered_properties / agent_property_authorizations writes + * (migration 026) still happen — dual-write for one release as a fallback + * before PR 5 drops the old tables. + * + * Closes the gap surfaced by Setupad escalation #218: properties that landed + * in discovered_properties via the crawler never made it into the catalog + * (migration 336 was a one-time seed). With this PR, every successful crawl + * lands in both places. + */ +import { describe, it, expect, beforeAll, beforeEach, afterAll } from 'vitest'; +import { initializeDatabase, closeDatabase } from '../../src/db/client.js'; +import { runMigrations } from '../../src/db/migrate.js'; +import { PublisherDatabase } from '../../src/db/publisher-db.js'; +import { FederatedIndexService } from '../../src/federated-index.js'; +import type { Pool } from 'pg'; + +const TEST_DOMAIN = 'crawler-cache.example.com'; +const TEST_AGENT = 'https://agent.crawler-cache.example.com/mcp'; +// Cross-publisher fixtures used by the tenant-isolation tests. +const VICTIM_DOMAIN = 'victim.crawler-cache.example.com'; +const ATTACKER_DOMAIN = 'attacker.crawler-cache.example.com'; + +const FIXTURE_MANIFEST = { + $schema: 'https://adcontextprotocol.org/schemas/v2/adagents.json', + authorized_agents: [ + { + url: TEST_AGENT, + authorized_for: 'Display inventory across all properties', + property_ids: ['site_main', 'app_ios'], + }, + ], + properties: [ + { + property_id: 'site_main', + property_type: 'website', + name: 'Crawler Cache Main Site', + identifiers: [ + { type: 'domain', value: TEST_DOMAIN }, + { type: 'subdomain', value: `news.${TEST_DOMAIN}` }, + ], + tags: ['flagship'], + }, + { + property_id: 'app_ios', + property_type: 'mobile_app', + name: 'Crawler Cache iOS App', + identifiers: [{ type: 'ios_bundle', value: 'com.example.crawlercache' }], + }, + ], + last_updated: '2026-04-25T00:00:00Z', +}; + +describe('Registry crawler cache (PR 2 of #3177)', () => { + let pool: Pool; + let publisherDb: PublisherDatabase; + let federatedIndex: FederatedIndexService; + + beforeAll(async () => { + pool = initializeDatabase({ + connectionString: process.env.DATABASE_URL || 'postgresql://adcp:localdev@localhost:5432/adcp_test', + }); + await runMigrations(); + publisherDb = new PublisherDatabase(); + federatedIndex = new FederatedIndexService(); + }); + + // Scope cleanup tightly so parallel runs of other tests sharing the + // .example.com pattern don't trample our fixtures. + const TEST_CREATED_BY = [ + `adagents_json:${TEST_DOMAIN}`, + `adagents_json:${VICTIM_DOMAIN}`, + `adagents_json:${ATTACKER_DOMAIN}`, + 'test:tenant-isolation-seed', + ]; + const TEST_DOMAINS = [TEST_DOMAIN, VICTIM_DOMAIN, ATTACKER_DOMAIN]; + + async function clearTestFixtures() { + // Identifiers must clear before properties — catalog_identifiers FKs to + // catalog_properties. Delete via the property_rid join so any identifier + // value (including ones the tests didn't list explicitly) gets caught. + await pool.query( + `DELETE FROM catalog_identifiers + WHERE property_rid IN ( + SELECT property_rid FROM catalog_properties + WHERE created_by = ANY($1::text[]) + )`, + [TEST_CREATED_BY] + ); + await pool.query( + `DELETE FROM catalog_properties WHERE created_by = ANY($1::text[])`, + [TEST_CREATED_BY] + ); + await pool.query( + `DELETE FROM publishers WHERE domain = ANY($1::text[])`, + [TEST_DOMAINS] + ); + await pool.query( + 'DELETE FROM agent_property_authorizations WHERE agent_url = $1', + [TEST_AGENT] + ); + await pool.query( + `DELETE FROM discovered_properties WHERE publisher_domain = ANY($1::text[])`, + [TEST_DOMAINS] + ); + await pool.query('DELETE FROM discovered_agents WHERE agent_url = $1', [TEST_AGENT]); + await pool.query( + `DELETE FROM agent_publisher_authorizations WHERE publisher_domain = ANY($1::text[])`, + [TEST_DOMAINS] + ); + } + + beforeEach(async () => { + await clearTestFixtures(); + }); + + afterAll(async () => { + await clearTestFixtures(); + await closeDatabase(); + }); + + describe('publishers cache', () => { + it('upserts publishers row with adagents_json source_type and manifest body', async () => { + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + + const { rows } = await pool.query<{ + domain: string; + adagents_json: unknown; + source_type: string; + last_validated: Date | null; + }>( + `SELECT domain, adagents_json, source_type, last_validated + FROM publishers WHERE domain = $1`, + [TEST_DOMAIN] + ); + + expect(rows).toHaveLength(1); + expect(rows[0].source_type).toBe('adagents_json'); + expect(rows[0].last_validated).not.toBeNull(); + + // The manifest body comes back parsed (JSONB), not as a string. + const stored = rows[0].adagents_json as typeof FIXTURE_MANIFEST; + expect(stored.authorized_agents).toEqual(FIXTURE_MANIFEST.authorized_agents); + expect(stored.properties).toHaveLength(2); + expect(stored.last_updated).toBe('2026-04-25T00:00:00Z'); + }); + + it('preserves org/ownership metadata on re-crawl (ON CONFLICT semantics)', async () => { + // Seed a row that was registered by an org BEFORE the crawler runs. + await pool.query( + `INSERT INTO publishers (domain, source_type, workos_organization_id, created_by_email) + VALUES ($1, 'community', 'org_test_publisher_owner', 'owner@example.com')`, + [TEST_DOMAIN] + ); + + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + + const { rows } = await pool.query<{ + source_type: string; + workos_organization_id: string | null; + created_by_email: string | null; + }>( + `SELECT source_type, workos_organization_id, created_by_email + FROM publishers WHERE domain = $1`, + [TEST_DOMAIN] + ); + + expect(rows[0].source_type).toBe('adagents_json'); + expect(rows[0].workos_organization_id).toBe('org_test_publisher_owner'); + expect(rows[0].created_by_email).toBe('owner@example.com'); + }); + }); + + describe('catalog projection', () => { + it('materializes catalog_properties with adagents_url and authoritative source', async () => { + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + + const { rows } = await pool.query<{ + property_rid: string; + property_id: string | null; + classification: string; + source: string; + status: string; + adagents_url: string | null; + }>( + `SELECT property_rid, property_id, classification, source, status, adagents_url + FROM catalog_properties + WHERE created_by = $1 + ORDER BY property_id NULLS LAST`, + [`adagents_json:${TEST_DOMAIN}`] + ); + + expect(rows).toHaveLength(2); + const ids = rows.map((r) => r.property_id); + expect(ids).toContain('site_main'); + expect(ids).toContain('app_ios'); + for (const row of rows) { + expect(row.classification).toBe('property'); + expect(row.source).toBe('authoritative'); + expect(row.status).toBe('active'); + expect(row.adagents_url).toBe(`https://${TEST_DOMAIN}/.well-known/adagents.json`); + } + }); + + it('materializes catalog_identifiers with evidence=adagents_json and confidence=authoritative', async () => { + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + + const { rows } = await pool.query<{ + identifier_type: string; + identifier_value: string; + evidence: string; + confidence: string; + }>( + `SELECT identifier_type, identifier_value, evidence, confidence + FROM catalog_identifiers + WHERE identifier_value IN ($1, $2, $3) + ORDER BY identifier_value`, + [TEST_DOMAIN, `news.${TEST_DOMAIN}`, 'com.example.crawlercache'] + ); + + expect(rows).toHaveLength(3); + for (const row of rows) { + expect(row.evidence).toBe('adagents_json'); + expect(row.confidence).toBe('authoritative'); + } + const valuesByType = new Map(rows.map((r) => [r.identifier_value, r.identifier_type])); + expect(valuesByType.get(TEST_DOMAIN)).toBe('domain'); + expect(valuesByType.get(`news.${TEST_DOMAIN}`)).toBe('subdomain'); + expect(valuesByType.get('com.example.crawlercache')).toBe('ios_bundle'); + }); + + it('normalizes identifier values to lowercase before catalog insert', async () => { + // catalog_identifiers has a chk_identifier_lowercase CHECK; the writer + // must run normalizeIdentifier so the row inserts cleanly even when the + // publisher's adagents.json declares a mixed-case value. + const mixedCaseManifest = { + ...FIXTURE_MANIFEST, + properties: [ + { + property_id: 'site_main', + property_type: 'website', + name: 'Mixed Case Site', + identifiers: [{ type: 'ios_bundle', value: 'COM.EXAMPLE.CRAWLERCACHE' }], + }, + ], + }; + + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: mixedCaseManifest }); + + const { rows } = await pool.query<{ identifier_value: string }>( + `SELECT identifier_value FROM catalog_identifiers + WHERE identifier_value = 'com.example.crawlercache'` + ); + expect(rows).toHaveLength(1); + }); + + it('lowercases rss_url path so chk_identifier_lowercase doesn\'t silently roll back', async () => { + // normalizeRssUrl preserves URL path case ("Feed.xml" stays mixed). Without + // the writer's defensive lowercase, this triggers a 23514 check_violation + // mid-transaction and the entire crawl is silently rolled back. + const rssManifest = { + ...FIXTURE_MANIFEST, + properties: [ + { + property_id: 'feed_main', + property_type: 'podcast', + name: 'Crawler Cache Feed', + identifiers: [{ type: 'rss_url', value: `https://${TEST_DOMAIN}/Feed.xml` }], + }, + ], + }; + + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: rssManifest }); + + const { rows } = await pool.query<{ identifier_value: string }>( + `SELECT identifier_value FROM catalog_identifiers + WHERE identifier_type = 'rss_url' AND property_rid IN ( + SELECT property_rid FROM catalog_properties WHERE created_by = $1 + )`, + [`adagents_json:${TEST_DOMAIN}`] + ); + expect(rows).toHaveLength(1); + expect(rows[0].identifier_value).toBe(`https://${TEST_DOMAIN}/feed.xml`); + }); + + it('reuses property_rid on re-crawl rather than forking identity', async () => { + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + const first = await pool.query<{ property_rid: string }>( + `SELECT property_rid FROM catalog_identifiers + WHERE identifier_type = 'domain' AND identifier_value = $1`, + [TEST_DOMAIN] + ); + expect(first.rows).toHaveLength(1); + const ridAfterFirstCrawl = first.rows[0].property_rid; + + // Second crawl with the same manifest: no new catalog_properties row, + // and the existing identifier still points at the same rid. + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + + const second = await pool.query<{ property_rid: string }>( + `SELECT property_rid FROM catalog_identifiers + WHERE identifier_type = 'domain' AND identifier_value = $1`, + [TEST_DOMAIN] + ); + expect(second.rows[0].property_rid).toBe(ridAfterFirstCrawl); + + const propCount = await pool.query<{ c: string }>( + `SELECT count(*)::text AS c FROM catalog_properties + WHERE created_by = $1`, + [`adagents_json:${TEST_DOMAIN}`] + ); + expect(propCount.rows[0].c).toBe('2'); + }); + }); + + describe('tenant isolation', () => { + it('refuses to rebind a victim\'s identifier when another publisher claims it', async () => { + // Victim claims its own domain. + await publisherDb.upsertAdagentsCache({ + domain: VICTIM_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'victim_site', + property_type: 'website', + name: 'Victim Site', + identifiers: [{ type: 'domain', value: VICTIM_DOMAIN }], + }, + ], + }, + }); + + const beforeAttacker = await pool.query<{ property_rid: string; created_by: string | null }>( + `SELECT cp.property_rid, cp.created_by + FROM catalog_identifiers ci + JOIN catalog_properties cp ON cp.property_rid = ci.property_rid + WHERE ci.identifier_type = 'domain' AND ci.identifier_value = $1`, + [VICTIM_DOMAIN] + ); + expect(beforeAttacker.rows).toHaveLength(1); + const victimRid = beforeAttacker.rows[0].property_rid; + expect(beforeAttacker.rows[0].created_by).toBe(`adagents_json:${VICTIM_DOMAIN}`); + + // Attacker publishes a manifest naming the victim's domain alongside its own. + await publisherDb.upsertAdagentsCache({ + domain: ATTACKER_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'attacker_bundle', + property_type: 'website', + name: 'Attacker Bundle', + identifiers: [ + { type: 'domain', value: VICTIM_DOMAIN }, + { type: 'domain', value: ATTACKER_DOMAIN }, + ], + }, + ], + }, + }); + + // Victim's identifier still points at the victim's rid; not rebound. + const victimIdentifier = await pool.query<{ property_rid: string }>( + `SELECT property_rid FROM catalog_identifiers + WHERE identifier_type = 'domain' AND identifier_value = $1`, + [VICTIM_DOMAIN] + ); + expect(victimIdentifier.rows[0].property_rid).toBe(victimRid); + + // Attacker's own identifier was NOT bound to the victim's rid (refusal + // skipped the whole projection — neither side of the merge lands). + const attackerIdentifier = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM catalog_identifiers + WHERE identifier_type = 'domain' AND identifier_value = $1`, + [ATTACKER_DOMAIN] + ); + expect(attackerIdentifier.rows[0].count).toBe('0'); + + // Victim's catalog property is still authored by the victim, not the attacker. + const victimProperty = await pool.query<{ created_by: string | null; adagents_url: string | null }>( + `SELECT created_by, adagents_url FROM catalog_properties WHERE property_rid = $1`, + [victimRid] + ); + expect(victimProperty.rows[0].created_by).toBe(`adagents_json:${VICTIM_DOMAIN}`); + expect(victimProperty.rows[0].adagents_url).toBe( + `https://${VICTIM_DOMAIN}/.well-known/adagents.json` + ); + }); + + it('refuses to silently merge two existing properties when a manifest spans both', async () => { + // Seed two distinct properties from a non-adagents source (community/seed), + // each with its own identifier. A later manifest that names BOTH identifiers + // in one property would silently merge them without this guard. + await pool.query( + `INSERT INTO catalog_properties + (property_rid, property_id, classification, source, status, created_by) + VALUES + ('11111111-1111-7111-9111-111111111111', 'alpha', 'property', 'contributed', 'active', 'test:tenant-isolation-seed'), + ('22222222-2222-7222-9222-222222222222', 'beta', 'property', 'contributed', 'active', 'test:tenant-isolation-seed')`, + ); + await pool.query( + `INSERT INTO catalog_identifiers + (id, property_rid, identifier_type, identifier_value, evidence, confidence) + VALUES + (gen_random_uuid(), '11111111-1111-7111-9111-111111111111', 'ios_bundle', 'com.example.alpha', 'member_resolve', 'medium'), + (gen_random_uuid(), '22222222-2222-7222-9222-222222222222', 'ios_bundle', 'com.example.beta', 'member_resolve', 'medium')`, + ); + + await publisherDb.upsertAdagentsCache({ + domain: TEST_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'wants_to_merge', + property_type: 'mobile_app', + name: 'Tries To Merge Alpha and Beta', + identifiers: [ + { type: 'ios_bundle', value: 'com.example.alpha' }, + { type: 'ios_bundle', value: 'com.example.beta' }, + ], + }, + ], + }, + }); + + // Both seed properties survive untouched; no third property was minted + // for this manifest's claim. + const seedRids = await pool.query<{ property_rid: string }>( + `SELECT property_rid FROM catalog_identifiers + WHERE identifier_value IN ('com.example.alpha', 'com.example.beta') + ORDER BY identifier_value` + ); + expect(seedRids.rows.map((r) => r.property_rid).sort()).toEqual( + [ + '11111111-1111-7111-9111-111111111111', + '22222222-2222-7222-9222-222222222222', + ].sort() + ); + + const newProps = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM catalog_properties WHERE created_by = $1`, + [`adagents_json:${TEST_DOMAIN}`] + ); + expect(newProps.rows[0].count).toBe('0'); + }); + + it('refuses cross-publisher domain claims regardless of crawl ordering (land-grab)', async () => { + // Attacker crawls FIRST, claiming the victim's domain alongside its own + // (attacker-first ordering — distinct from the victim-first rebind case). + // Without the anchor rule the attacker would mint a rid pointing + // domain:VICTIM_DOMAIN at the attacker's adagents_url, and the victim's + // later crawl would be the one refused. + await publisherDb.upsertAdagentsCache({ + domain: ATTACKER_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'land_grab', + property_type: 'website', + name: 'Land Grab Attempt', + identifiers: [ + { type: 'domain', value: ATTACKER_DOMAIN }, + { type: 'domain', value: VICTIM_DOMAIN }, + ], + }, + ], + }, + }); + + // Anchor rule refuses the entire property — neither identifier lands. + const anyAttackerProp = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM catalog_properties + WHERE created_by = $1`, + [`adagents_json:${ATTACKER_DOMAIN}`] + ); + expect(anyAttackerProp.rows[0].count).toBe('0'); + + const victimIdentifier = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM catalog_identifiers + WHERE identifier_type = 'domain' AND identifier_value = $1`, + [VICTIM_DOMAIN] + ); + expect(victimIdentifier.rows[0].count).toBe('0'); + + // Victim's own crawl (which only declares its own anchored domain) lands cleanly. + await publisherDb.upsertAdagentsCache({ + domain: VICTIM_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'victim_site', + property_type: 'website', + name: 'Victim Site', + identifiers: [{ type: 'domain', value: VICTIM_DOMAIN }], + }, + ], + }, + }); + + const victimRid = await pool.query<{ property_rid: string; created_by: string | null }>( + `SELECT cp.property_rid, cp.created_by + FROM catalog_identifiers ci + JOIN catalog_properties cp ON cp.property_rid = ci.property_rid + WHERE ci.identifier_type = 'domain' AND ci.identifier_value = $1`, + [VICTIM_DOMAIN] + ); + expect(victimRid.rows).toHaveLength(1); + expect(victimRid.rows[0].created_by).toBe(`adagents_json:${VICTIM_DOMAIN}`); + }); + + it('refuses to take over a seed-source rid without a publisher-anchored identifier', async () => { + // Seed the catalog with a rid created by a non-adagents source (mimicking + // migration 336 or a hosted_properties seed) — adagents_url is NULL, so + // a COALESCE-based reuse path would happily bind the attacker's URL. + await pool.query( + `INSERT INTO catalog_properties + (property_rid, property_id, classification, source, status, adagents_url, created_by) + VALUES ('33333333-3333-7333-9333-333333333333', 'seeded', 'property', 'authoritative', 'active', NULL, 'test:tenant-isolation-seed')` + ); + await pool.query( + `INSERT INTO catalog_identifiers + (id, property_rid, identifier_type, identifier_value, evidence, confidence) + VALUES (gen_random_uuid(), '33333333-3333-7333-9333-333333333333', 'ios_bundle', 'com.example.victimapp', 'member_resolve', 'medium')` + ); + + // Attacker publishes a manifest claiming the seeded bundle ID with NO + // anchor identifier proving they're authoritative for it. Without the + // anchor rule the writer would reuse the seed rid and overwrite + // adagents_url via COALESCE(NULL, attacker_url). + await publisherDb.upsertAdagentsCache({ + domain: ATTACKER_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'unanchored_claim', + property_type: 'mobile_app', + name: 'Unanchored Bundle Claim', + identifiers: [{ type: 'ios_bundle', value: 'com.example.victimapp' }], + }, + ], + }, + }); + + // Seed rid's adagents_url is still NULL — attacker did not take over. + const seedRow = await pool.query<{ adagents_url: string | null; created_by: string | null }>( + `SELECT adagents_url, created_by FROM catalog_properties + WHERE property_rid = '33333333-3333-7333-9333-333333333333'` + ); + expect(seedRow.rows[0].adagents_url).toBeNull(); + expect(seedRow.rows[0].created_by).toBe('test:tenant-isolation-seed'); + }); + + it('lets a publisher adopt a seed-source rid when the manifest carries an anchor identifier', async () => { + // Same seed setup as above, but pre-link the publisher's own domain to + // the seed rid (modeling migration 336's case where a discovered_property + // had both a domain identifier and a bundle ID). + await pool.query( + `INSERT INTO catalog_properties + (property_rid, property_id, classification, source, status, adagents_url, created_by) + VALUES ('44444444-4444-7444-9444-444444444444', NULL, 'property', 'authoritative', 'active', NULL, 'test:tenant-isolation-seed')` + ); + await pool.query( + `INSERT INTO catalog_identifiers + (id, property_rid, identifier_type, identifier_value, evidence, confidence) + VALUES + (gen_random_uuid(), '44444444-4444-7444-9444-444444444444', 'domain', $1, 'adagents_json', 'authoritative'), + (gen_random_uuid(), '44444444-4444-7444-9444-444444444444', 'ios_bundle', 'com.example.alpha', 'member_resolve', 'medium')`, + [VICTIM_DOMAIN] + ); + + // Legitimate publisher (matching the seeded domain) crawls, declaring + // the same domain and the same bundle ID. The anchor proves authority, + // so the publisher takes ownership and adagents_url is set. + await publisherDb.upsertAdagentsCache({ + domain: VICTIM_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'adopt_seed', + property_type: 'website', + name: 'Adopt Seed', + identifiers: [ + { type: 'domain', value: VICTIM_DOMAIN }, + { type: 'ios_bundle', value: 'com.example.alpha' }, + ], + }, + ], + }, + }); + + const adopted = await pool.query<{ adagents_url: string | null; property_id: string | null }>( + `SELECT adagents_url, property_id FROM catalog_properties + WHERE property_rid = '44444444-4444-7444-9444-444444444444'` + ); + expect(adopted.rows[0].adagents_url).toBe( + `https://${VICTIM_DOMAIN}/.well-known/adagents.json` + ); + expect(adopted.rows[0].property_id).toBe('adopt_seed'); + }); + + it('does not abort the rest of the manifest when one property is refused', async () => { + // Pre-claim an identifier from a different publisher, so when our manifest + // tries to bind it the projection is refused for that property only. The + // other (clean) property in the same manifest should still land. + await publisherDb.upsertAdagentsCache({ + domain: ATTACKER_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'attacker_app', + property_type: 'mobile_app', + name: 'Attacker App', + identifiers: [{ type: 'ios_bundle', value: 'com.example.victimapp' }], + }, + ], + }, + }); + + await publisherDb.upsertAdagentsCache({ + domain: TEST_DOMAIN, + manifest: { + authorized_agents: [], + properties: [ + { + property_id: 'clean_site', + property_type: 'website', + name: 'Clean Site', + identifiers: [{ type: 'domain', value: TEST_DOMAIN }], + }, + { + property_id: 'collides', + property_type: 'mobile_app', + name: 'Collides With Attacker', + identifiers: [{ type: 'ios_bundle', value: 'com.example.victimapp' }], + }, + ], + }, + }); + + // Clean property landed. + const cleanProp = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM catalog_properties + WHERE created_by = $1 AND property_id = 'clean_site'`, + [`adagents_json:${TEST_DOMAIN}`] + ); + expect(cleanProp.rows[0].count).toBe('1'); + + // Collides property did NOT land (refused — attacker still owns the rid). + const collidesProp = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM catalog_properties + WHERE created_by = $1 AND property_id = 'collides'`, + [`adagents_json:${TEST_DOMAIN}`] + ); + expect(collidesProp.rows[0].count).toBe('0'); + + // Publishers cache for the original domain still updated (the manifest + // body cache write isn't gated on per-property success). + const cache = await pool.query<{ source_type: string }>( + `SELECT source_type FROM publishers WHERE domain = $1`, + [TEST_DOMAIN] + ); + expect(cache.rows[0].source_type).toBe('adagents_json'); + }); + }); + + describe('dual-write fallback to legacy tables', () => { + it('still writes discovered_properties and agent_property_authorizations alongside the new cache', async () => { + // Mirror what crawler.ts does for a successful adagents.json crawl: + // call the publisher cache writer AND the federated-index writer. + await publisherDb.upsertAdagentsCache({ domain: TEST_DOMAIN, manifest: FIXTURE_MANIFEST }); + + for (const authorizedAgent of FIXTURE_MANIFEST.authorized_agents) { + await federatedIndex.recordAgentFromAdagentsJson( + authorizedAgent.url, + TEST_DOMAIN, + authorizedAgent.authorized_for, + authorizedAgent.property_ids + ); + for (const prop of FIXTURE_MANIFEST.properties) { + await federatedIndex.recordProperty( + { + property_id: prop.property_id, + publisher_domain: TEST_DOMAIN, + property_type: prop.property_type, + name: prop.name, + identifiers: prop.identifiers, + tags: prop.tags, + }, + authorizedAgent.url, + authorizedAgent.authorized_for + ); + } + } + + // New tables + const pub = await pool.query<{ source_type: string }>( + `SELECT source_type FROM publishers WHERE domain = $1`, + [TEST_DOMAIN] + ); + expect(pub.rows[0].source_type).toBe('adagents_json'); + + // Legacy tables + const legacyProps = await pool.query<{ name: string; property_id: string | null }>( + `SELECT name, property_id FROM discovered_properties WHERE publisher_domain = $1 + ORDER BY property_id NULLS LAST`, + [TEST_DOMAIN] + ); + expect(legacyProps.rows).toHaveLength(2); + expect(legacyProps.rows.map((r) => r.property_id)).toEqual( + expect.arrayContaining(['site_main', 'app_ios']) + ); + + const legacyAuth = await pool.query<{ agent_url: string }>( + `SELECT apa.agent_url + FROM agent_property_authorizations apa + JOIN discovered_properties dp ON dp.id = apa.property_id + WHERE dp.publisher_domain = $1`, + [TEST_DOMAIN] + ); + expect(legacyAuth.rows.map((r) => r.agent_url)).toEqual( + expect.arrayContaining([TEST_AGENT]) + ); + }); + }); +});