fix(search): require the purge preflight's trigger to fire and call the retiring function

- embedding_search_sync must be enabled for ordinary writes (origin or always, not replica-only) and call public.sync_embedding_search
- Drop the redundant client port option; postgres.js reads the URL port

Author: waleedlatif1

packages/db/scripts/purge-search-projection.integration.ts

@@ -392,11 +392,24 @@ describe('Search candidate retirement in PostgreSQL', () => {
       await applyMigration()
       await sql`ALTER TABLE embedding DISABLE TRIGGER embedding_search_sync`
       expect(runPurge('--execute', '--pause-ms', '0').status).toBe(1)
+      // A replica-only trigger never fires for application writes.
+      await sql`ALTER TABLE embedding ENABLE REPLICA TRIGGER embedding_search_sync`
+      expect(runPurge('--execute', '--pause-ms', '0').status).toBe(1)
+      await sql`ALTER TABLE embedding ENABLE TRIGGER embedding_search_sync`
+
+      await sql.unsafe(`CREATE FUNCTION public.other_embedding_sync()
+        RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RETURN NEW; END; $$`)
+      await sql.unsafe(`CREATE OR REPLACE TRIGGER embedding_search_sync AFTER INSERT ON embedding
+        FOR EACH ROW EXECUTE FUNCTION public.other_embedding_sync()`)
+      expect(runPurge('--execute', '--pause-ms', '0').status).toBe(1)
       expect(await candidateCount('search')).toBe(3)
     } finally {
       await sql`ALTER TABLE embedding ENABLE TRIGGER embedding_search_sync`
+      await sql.unsafe(provisioned.sync.replace('CREATE TRIGGER', 'CREATE OR REPLACE TRIGGER'))
+      await sql`DROP FUNCTION IF EXISTS public.other_embedding_sync()`
       await applyMigration()
     }
+    expect((await definitions())[0].sync).toBe(provisioned.sync)
     const dryRun = runPurge()
     expect(dryRun.stdout).toContain('"projectionRetired":true')
     const execute = runPurge('--execute', '--pause-ms', '0')

packages/db/scripts/purge-search-projection.ts

@@ -134,7 +134,6 @@ export function connectOperatorDatabase(rawUrl: string | undefined): Sql {
   clientUrl.searchParams.delete('sslmode')
   clientUrl.searchParams.delete('sslrootcert')
   return postgres(clientUrl.toString(), {
-    port: Number(url.port || '5432'),
     ssl: hosted ? 'verify-full' : false,
     max: 1,
     prepare: false,
@@ -153,7 +152,7 @@ export function connectOperatorDatabase(rawUrl: string | undefined): Sql {
 
 /**
  * Whether migration 0449 is live: `public.sync_embedding_search` has the body that deletes a
- * Search chunk's candidate row, and its `embedding_search_sync` trigger is enabled. Before that,
+ * Search chunk's candidate row, and the enabled `embedding_search_sync` trigger calls it. Before that,
  * chunk writes keep projecting the rows a purge deletes.
  */
 async function isProjectionRetired(sql: Sql): Promise<boolean> {
@@ -162,7 +161,7 @@ async function isProjectionRetired(sql: Sql): Promise<boolean> {
       EXISTS (
         SELECT 1 FROM pg_trigger
         WHERE tgrelid = to_regclass('public.embedding') AND tgname = 'embedding_search_sync'
-          AND tgenabled <> 'D'
+          AND tgenabled IN ('O', 'A') AND tgfoid = to_regproc('public.sync_embedding_search')
       ) AS trigger_enabled`
   return (
     live.trigger_enabled && live.source !== null && isCurrentSyncEmbeddingSearchSource(live.source)