Function applyOrg

scripts/migration/service-split-postgres-cli.ts:235–488

(sql: SqlClient, org: OrgPlan)

  `${operationStorageKey(integration, "").split(".").slice(0, 2).join(".")}.%`;

export const applyOrg = async (sql: SqlClient, org: OrgPlan): Promise<void> => {
  if (org.completed) return;
  if (org.hardErrors.length > 0) {
    throw new Error(`org ${org.tenantHash} has unresolved hard errors; apply is blocked`);
  }
  await sql.begin(async (tx) => {
    const now = new Date();
    for (const copy of org.blobCopies.filter((item) => item.backend === "database")) {
      const [source] = await tx.unsafe<{ readonly value: string }[]>(
        "SELECT value FROM blob WHERE namespace = $1 AND key = $2 LIMIT 1",
        [copy.sourceNamespace, copy.key],
      );
      if (!source) {
        throw new Error(`Missing source blob ${copy.sourceNamespace}/${copy.key}`);
      }
      await tx.unsafe(
        `
          INSERT INTO blob (namespace, key, value, row_id, id)
          VALUES ($1, $2, $3, $4, $5)
          ON CONFLICT (id) DO NOTHING
        `,
        [
          copy.targetNamespace,
          copy.key,
          source.value,
          stableId("blob", org.tenant, copy.targetNamespace, copy.key),
          JSON.stringify([copy.targetNamespace, copy.key]),
        ],
      );
    }

    for (const row of org.integrations.filter((item) => item.action === "create")) {
      await tx.unsafe(
        `
          INSERT INTO integration (
            slug, plugin_id, name, description, config, health_check, config_revised_at,
            can_remove, can_refresh, created_at, updated_at, row_id, tenant
          )
          VALUES ($1, $2, $3, $4, $5::json, $6::json, NULL, true, true, $7, $7, $8, $9)
          ON CONFLICT (tenant, slug) DO NOTHING
        `,
        [
          row.target.slug,
          row.target.pluginId,
          row.target.name,
          row.target.description,
          JSON.stringify(scrubJson(row.config)),
          row.healthCheck ? JSON.stringify(scrubJson(row.healthCheck)) : null,
          now,
          stableId("integration", org.tenant, row.target.slug),
          org.tenant,
        ],
      );
    }

    for (const row of org.connections.filter((item) => item.action === "clone")) {
      await tx.unsafe(
        `