feat(registry): swap property readers to catalog UNION (PR 4a of #3177)

.changeset/registry-property-readers-catalog.md

@@ -0,0 +1,23 @@
+---
+---
+
+Property-side readers in `federated-index-db.ts` and `property-db.ts` now
+union the catalog-side `publishers` cache (PR 1 of #3177) with the legacy
+`discovered_properties` and `discovered_publishers` tables. Crawl-sourced
+properties that landed via the new writer path but missed the legacy
+table — gatavo.com surfaced via Setupad escalation #218 — now appear on
+the registry surfaces alongside legacy data.
+
+Affects `hasValidAdagents`, `getPropertiesForDomain`,
+`getDiscoveredPropertiesByDomain`, `getAllPropertiesForRegistry`, and
+`getPropertyRegistryStats`. Authorization-side readers
+(`getPropertiesForAgent`, `findAgentsForPropertyIdentifier`,
+`getPublisherDomainsForAgent`) stay on legacy tables — those need the
+authorization model decision and ship in PR 4b.
+
+Legacy wins on collisions so callers that hold a
+`discovered_properties.id` keep dereferencing it correctly during the
+dual-write window. After PR 5 drops the legacy tables the legacy half
+of the union goes away.
+
+Refs #3177. Builds on #3195 / #3218 / #3221.

server/src/db/federated-index-db.ts

@@ -155,19 +155,33 @@ export class FederatedIndexDatabase {
   }
 
   /**
-   * Check whether a publisher domain has a valid adagents.json (from crawl data).
-   * Returns true if any record for this domain has has_valid_adagents = true,
-   * null if the domain has never been discovered.
+   * Check whether a publisher domain has a valid adagents.json.
+   *
+   * Reads from both the catalog-side `publishers` overlay (PR 1 of #3177)
+   * and the legacy `discovered_publishers` table during the dual-write
+   * window. A presence in `publishers` with `source_type='adagents_json'`
+   * means the crawler successfully validated and cached the file — that
+   * always wins. Otherwise fall back to bool_or over discovered_publishers
+   * so the historical three-state contract (true / false / null) is
+   * preserved when only the legacy path has data.
    */
   async hasValidAdagents(domain: string): Promise<boolean | null> {
-    const result = await query<{ has_valid: boolean }>(
-      `SELECT bool_or(has_valid_adagents) as has_valid
-       FROM discovered_publishers
-       WHERE domain = $1`,
+    const result = await query<{ catalog_present: boolean; legacy_or: boolean | null }>(
+      `SELECT
+         EXISTS(
+           SELECT 1 FROM publishers
+            WHERE domain = $1 AND source_type = 'adagents_json'
+         ) AS catalog_present,
+         (SELECT bool_or(has_valid_adagents)
+            FROM discovered_publishers
+           WHERE domain = $1) AS legacy_or`,
       [domain]
     );
-    if (!result.rows[0] || result.rows[0].has_valid === null) return null;
-    return result.rows[0].has_valid;
+    const row = result.rows[0];
+    if (!row) return null;
+    if (row.catalog_present) return true;
+    if (row.legacy_or === null) return null;
+    return row.legacy_or;
   }
 
   /**
@@ -423,13 +437,68 @@ export class FederatedIndexDatabase {
   }
 
   /**
-   * Get all properties for a publisher domain
+   * Get all properties for a publisher domain.
+   *
+   * UNION over the legacy `discovered_properties` table and the catalog-side
+   * `publishers.adagents_json` JSONB during the dual-write window of #3177.
+   * The legacy half wins on (publisher_domain, name, property_type)
+   * collisions so callers that hold a `discovered_properties.id` keep
+   * dereferencing it correctly. Catalog-only rows surface for properties
+   * that landed via the new writer path but didn't get a legacy row (e.g.
+   * post-seed gatavo.com / Setupad #218). After PR 5 the legacy half is
+   * removed and the query collapses to catalog-only.
    */
   async getPropertiesForDomain(domain: string): Promise<DiscoveredProperty[]> {
     const result = await query<DiscoveredProperty>(
-      `SELECT * FROM discovered_properties
-       WHERE publisher_domain = $1
-       ORDER BY property_type, name`,
+      `WITH unioned AS (
+         SELECT id, property_id, publisher_domain, property_type, name,
+                identifiers, tags, discovered_at, last_validated, expires_at,
+                0 AS src_priority
+           FROM discovered_properties
+          WHERE publisher_domain = $1
+         UNION ALL
+         SELECT
+           cp.property_rid AS id,
+           prop->>'property_id' AS property_id,
+           p.domain AS publisher_domain,
+           prop->>'property_type' AS property_type,
+           prop->>'name' AS name,
+           CASE WHEN jsonb_typeof(prop->'identifiers') = 'array'
+                THEN prop->'identifiers'
+                ELSE '[]'::jsonb END AS identifiers,
+           COALESCE(
+             ARRAY(SELECT jsonb_array_elements_text(
+               CASE WHEN jsonb_typeof(prop->'tags') = 'array'
+                    THEN prop->'tags'
+                    ELSE '[]'::jsonb END
+             )),
+             ARRAY[]::text[]
+           ) AS tags,
+           cp.created_at AS discovered_at,
+           p.last_validated AS last_validated,
+           p.expires_at AS expires_at,
+           1 AS src_priority
+           FROM publishers p
+          CROSS JOIN LATERAL jsonb_array_elements(
+            CASE WHEN jsonb_typeof(p.adagents_json->'properties') = 'array'
+                 THEN p.adagents_json->'properties'
+                 ELSE '[]'::jsonb END
+          ) AS prop
+           LEFT JOIN catalog_properties cp
+                  ON cp.property_id = prop->>'property_id'
+                 AND cp.created_by = 'adagents_json:' || p.domain
+          WHERE p.domain = $1
+            AND p.source_type = 'adagents_json'
+            AND prop->>'name' IS NOT NULL
+            AND prop->>'property_type' IS NOT NULL
+       ), deduped AS (
+         SELECT DISTINCT ON (publisher_domain, name, property_type)
+                id, property_id, publisher_domain, property_type, name,
+                identifiers, tags, discovered_at, last_validated, expires_at
+           FROM unioned
+          ORDER BY publisher_domain, name, property_type, src_priority
+       )
+       SELECT * FROM deduped ORDER BY property_type, name`,
       [domain]
     );
     return result.rows.map(row => this.deserializeProperty(row));

server/src/db/migrations/433_catalog_adagents_lookup_index.sql

@@ -0,0 +1,16 @@
+-- Index supporting the property-side reader cutover (PR 4a of #3177).
+--
+-- The new readers in federated-index-db.ts and property-db.ts join
+-- catalog_properties on (created_by = 'adagents_json:<domain>',
+-- property_id = <manifest_property_id>) inside a CROSS JOIN LATERAL over
+-- the publisher's manifest. Without this index, each manifest property
+-- triggers a sequential scan of catalog_properties — O(N×M) for a
+-- publisher with M properties against N catalog rows.
+--
+-- The partial WHERE clause keeps the index narrow: only adagents-sourced
+-- rows ever match this lookup pattern. Community/system/seed rows are
+-- still served by their own classification/status indexes.
+
+CREATE INDEX IF NOT EXISTS idx_catalog_properties_adagents_lookup
+  ON catalog_properties (created_by, property_id)
+  WHERE created_by LIKE 'adagents_json:%';

server/src/db/property-db.ts

@@ -211,7 +211,13 @@ export class PropertyDatabase {
   // ========== Discovered Properties ==========
 
   /**
-   * Get discovered properties by publisher domain
+   * Get discovered properties by publisher domain.
+   *
+   * UNION over the legacy `discovered_properties` table and properties
+   * extracted from `publishers.adagents_json` JSONB. Legacy wins on
+   * collisions during the #3177 dual-write window so consumers that hold a
+   * `discovered_properties.id` keep working; catalog-only rows surface for
+   * post-seed properties that never landed in the legacy table.
    */
   async getDiscoveredPropertiesByDomain(domain: string): Promise<Array<{
     id: string;
@@ -233,7 +239,50 @@ export class PropertyDatabase {
       tags: string[];
       source_type: string;
     }>(
-      'SELECT * FROM discovered_properties WHERE publisher_domain = $1',
+      `WITH unioned AS (
+         SELECT id, property_id, publisher_domain, property_type, name,
+                identifiers, tags, source_type, 0 AS src_priority
+           FROM discovered_properties
+          WHERE publisher_domain = $1
+         UNION ALL
+         SELECT
+           cp.property_rid AS id,
+           prop->>'property_id' AS property_id,
+           p.domain AS publisher_domain,
+           prop->>'property_type' AS property_type,
+           prop->>'name' AS name,
+           CASE WHEN jsonb_typeof(prop->'identifiers') = 'array'
+                THEN prop->'identifiers'
+                ELSE '[]'::jsonb END AS identifiers,
+           COALESCE(
+             ARRAY(SELECT jsonb_array_elements_text(
+               CASE WHEN jsonb_typeof(prop->'tags') = 'array'
+                    THEN prop->'tags'
+                    ELSE '[]'::jsonb END
+             )),
+             ARRAY[]::text[]
+           ) AS tags,
+           'adagents_json'::text AS source_type,
+           1 AS src_priority
+           FROM publishers p
+          CROSS JOIN LATERAL jsonb_array_elements(
+            CASE WHEN jsonb_typeof(p.adagents_json->'properties') = 'array'
+                 THEN p.adagents_json->'properties'
+                 ELSE '[]'::jsonb END
+          ) AS prop
+           LEFT JOIN catalog_properties cp
+                  ON cp.property_id = prop->>'property_id'
+                 AND cp.created_by = 'adagents_json:' || p.domain
+          WHERE p.domain = $1
+            AND p.source_type = 'adagents_json'
+            AND prop->>'name' IS NOT NULL
+            AND prop->>'property_type' IS NOT NULL
+       )
+       SELECT DISTINCT ON (publisher_domain, name, property_type)
+              id, property_id, publisher_domain, property_type, name,
+              identifiers, tags, source_type
+         FROM unioned
+        ORDER BY publisher_domain, name, property_type, src_priority`,
       [domain.toLowerCase()]
     );
     return result.rows.map((row) => ({
@@ -271,7 +320,16 @@ export class PropertyDatabase {
   // ========== Property Registry (Combined View) ==========
 
   /**
-   * Get all properties (hosted + discovered) for registry view
+   * Get all properties (hosted + discovered) for registry view.
+   *
+   * Three-way UNION over hosted_properties (community/enriched), the legacy
+   * discovered_properties table, and the catalog-side publishers cache
+   * (PR 1 of #3177). A domain only surfaces once per query — hosted wins
+   * over crawl, and within the crawl side legacy wins over catalog so
+   * `agent_count` keeps being computed from the existing
+   * agent_property_authorizations join during the dual-write window. After
+   * PR 5 the legacy crawl half is removed and the catalog branch alone
+   * answers crawl-sourced rows.
    */
   async getAllPropertiesForRegistry(options: ListPropertiesOptions = {}): Promise<Array<{
     domain: string;
@@ -293,34 +351,64 @@ export class PropertyDatabase {
       verified: boolean;
     }>(
       `
-      -- Hosted properties
-      SELECT
-        publisher_domain as domain,
-        COALESCE(source_type, 'hosted') as source,
-        COALESCE(jsonb_array_length(adagents_json->'properties'), 0)::int as property_count,
-        COALESCE(jsonb_array_length(adagents_json->'authorized_agents'), 0)::int as agent_count,
-        domain_verified as verified
-      FROM hosted_properties
-      WHERE is_public = true
-        AND (review_status IS NULL OR review_status = 'approved')
-        AND ($1::text IS NULL OR publisher_domain ILIKE $1)
-
-      UNION ALL
-
-      -- Discovered properties (from crawled adagents.json)
-      SELECT
-        publisher_domain as domain,
-        CASE WHEN source_type = 'adagents_json' OR source_type IS NULL THEN 'adagents_json' ELSE 'discovered' END as source,
-        COUNT(*)::int as property_count,
-        (SELECT COUNT(DISTINCT apa.agent_url) FROM agent_property_authorizations apa
-         JOIN discovered_properties dp2 ON apa.property_id = dp2.id
-         WHERE dp2.publisher_domain = discovered_properties.publisher_domain)::int as agent_count,
-        true as verified
-      FROM discovered_properties
-      WHERE ($1::text IS NULL OR publisher_domain ILIKE $1)
-        AND publisher_domain NOT IN (SELECT publisher_domain FROM hosted_properties WHERE is_public = true)
-      GROUP BY publisher_domain, source_type
-
+      WITH hosted AS (
+        SELECT
+          publisher_domain as domain,
+          COALESCE(source_type, 'hosted') as source,
+          (CASE WHEN jsonb_typeof(adagents_json->'properties') = 'array'
+                THEN jsonb_array_length(adagents_json->'properties')
+                ELSE 0 END)::int as property_count,
+          (CASE WHEN jsonb_typeof(adagents_json->'authorized_agents') = 'array'
+                THEN jsonb_array_length(adagents_json->'authorized_agents')
+                ELSE 0 END)::int as agent_count,
+          domain_verified as verified,
+          0 as src_priority
+        FROM hosted_properties
+        WHERE is_public = true
+          AND (review_status IS NULL OR review_status = 'approved')
+          AND ($1::text IS NULL OR publisher_domain ILIKE $1)
+      ),
+      legacy AS (
+        SELECT
+          publisher_domain as domain,
+          CASE WHEN source_type = 'adagents_json' OR source_type IS NULL THEN 'adagents_json' ELSE 'discovered' END as source,
+          COUNT(*)::int as property_count,
+          (SELECT COUNT(DISTINCT apa.agent_url) FROM agent_property_authorizations apa
+           JOIN discovered_properties dp2 ON apa.property_id = dp2.id
+           WHERE dp2.publisher_domain = discovered_properties.publisher_domain)::int as agent_count,
+          true as verified,
+          1 as src_priority
+        FROM discovered_properties
+        WHERE ($1::text IS NULL OR publisher_domain ILIKE $1)
+          AND publisher_domain NOT IN (SELECT domain FROM hosted)
+        GROUP BY publisher_domain, source_type
+      ),
+      catalog_only AS (
+        SELECT
+          p.domain as domain,
+          'adagents_json'::text as source,
+          (CASE WHEN jsonb_typeof(p.adagents_json->'properties') = 'array'
+                THEN jsonb_array_length(p.adagents_json->'properties')
+                ELSE 0 END)::int as property_count,
+          (CASE WHEN jsonb_typeof(p.adagents_json->'authorized_agents') = 'array'
+                THEN jsonb_array_length(p.adagents_json->'authorized_agents')
+                ELSE 0 END)::int as agent_count,
+          true as verified,
+          2 as src_priority
+        FROM publishers p
+        WHERE p.source_type = 'adagents_json'
+          AND ($1::text IS NULL OR p.domain ILIKE $1)
+          AND p.domain NOT IN (SELECT domain FROM hosted)
+          AND p.domain NOT IN (SELECT domain FROM legacy)
+      )
+      SELECT domain, source, property_count, agent_count, verified
+      FROM (
+        SELECT * FROM hosted
+        UNION ALL
+        SELECT * FROM legacy
+        UNION ALL
+        SELECT * FROM catalog_only
+      ) all_sources
       ORDER BY domain
       LIMIT $2 OFFSET $3
       `,
@@ -331,30 +419,48 @@ export class PropertyDatabase {
   }
 
   /**
-   * Get aggregated stats for the property registry (counts by source type)
+   * Get aggregated stats for the property registry (counts by source type).
+   *
+   * Mirrors getAllPropertiesForRegistry's three-way UNION (hosted + legacy
+   * discovered + catalog-only publishers) so a domain that has both a
+   * legacy crawl row and a catalog row is counted exactly once. Hosted
+   * wins over crawl; legacy wins over catalog within the crawl side.
    */
   async getPropertyRegistryStats(search?: string): Promise<Record<string, number>> {
     const escapedSearch = search ? search.replace(/[%_\\]/g, '\\$&') : null;
     const searchParam = escapedSearch ? `%${escapedSearch}%` : null;
 
     const result = await query<{ source: string; count: number }>(
       `
-      SELECT source, COUNT(*)::int as count FROM (
-        -- Hosted properties
-        SELECT COALESCE(source_type, 'hosted') as source
+      WITH hosted AS (
+        SELECT publisher_domain AS domain, COALESCE(source_type, 'hosted') as source
         FROM hosted_properties
         WHERE is_public = true
           AND (review_status IS NULL OR review_status = 'approved')
           AND ($1::text IS NULL OR publisher_domain ILIKE $1)
-
+      ),
+      legacy AS (
+        SELECT publisher_domain AS domain,
+               CASE WHEN source_type = 'adagents_json' OR source_type IS NULL THEN 'adagents_json' ELSE 'discovered' END as source
+          FROM discovered_properties
+         WHERE ($1::text IS NULL OR publisher_domain ILIKE $1)
+           AND publisher_domain NOT IN (SELECT domain FROM hosted)
+         GROUP BY publisher_domain, source_type
+      ),
+      catalog_only AS (
+        SELECT p.domain AS domain, 'adagents_json'::text AS source
+          FROM publishers p
+         WHERE p.source_type = 'adagents_json'
+           AND ($1::text IS NULL OR p.domain ILIKE $1)
+           AND p.domain NOT IN (SELECT domain FROM hosted)
+           AND p.domain NOT IN (SELECT domain FROM legacy)
+      )
+      SELECT source, COUNT(*)::int AS count FROM (
+        SELECT source FROM hosted
         UNION ALL
-
-        -- Discovered properties
-        SELECT CASE WHEN source_type = 'adagents_json' OR source_type IS NULL THEN 'adagents_json' ELSE 'discovered' END as source
-        FROM discovered_properties
-        WHERE ($1::text IS NULL OR publisher_domain ILIKE $1)
-          AND publisher_domain NOT IN (SELECT publisher_domain FROM hosted_properties WHERE is_public = true)
-        GROUP BY publisher_domain, source_type
+        SELECT source FROM legacy
+        UNION ALL
+        SELECT source FROM catalog_only
       ) sub
       GROUP BY source
       `,