spb/datacenterindex

1/** entity_keys helpers (connector-scoped stable keys → entity ids), with merged_into following for facilities. */2import { sql } from "@dci/db";3import type { IngestContext, Tx } from "./common.js";45export async function entityIdForKey(tx: Tx, key: string, entityType: string): Promise<string | null> {6  const r = await tx.execute(sql`select entity_id from entity_keys where key = ${key} and entity_type = ${entityType} limit 1`);7  return r[0] ? String(r[0].entity_id) : null;8}910/** Facility id for a connector key, following `merged_into` chains (max 5 hops). Cached per batch. */11export async function facilityIdForKey(tx: Tx, ctx: IngestContext, key: string): Promise<string | null> {12  const cached = ctx.caches.facilityIdByKey.get(key);13  if (cached) return cached;14  let id: string | null = await entityIdForKey(tx, key, "facility");15  if (!id) return null;16  for (let hop = 0; hop < 5; hop++) {17    const rows: Array<Record<string, unknown>> = await tx.execute(sql`select merged_into from facilities where id = ${id}`);18    const m: unknown = rows[0]?.merged_into;19    if (!m) break;20    id = String(m);21  }22  ctx.caches.facilityIdByKey.set(key, id);23  return id;24}2526export async function upsertKey(tx: Tx, ctx: IngestContext, key: string, entityType: string, entityId: string): Promise<void> {27  await tx.execute(sql`insert into entity_keys (key, entity_type, entity_id, connector_id) values (${key}, ${entityType}, ${entityId}, ${ctx.run.connectorId})28    on conflict (key, entity_type) do update set entity_id = excluded.entity_id, connector_id = excluded.connector_id`);29  if (entityType === "facility") ctx.caches.facilityIdByKey.set(key, entityId);30}31