Function connectionsCreate

packages/core/sdk/src/executor.ts:3805–4386

(
      input: CreateConnectionInput,
    )

    // ------------------------------------------------------------------

    const connectionsCreate = (
      input: CreateConnectionInput,
    ): Effect.Effect<
      Connection,
      | IntegrationNotFoundError
      | ConnectionAlreadyExistsError
      | CredentialProviderNotRegisteredError
      | InvalidConnectionInputError
      | OrgWriteDeniedError
      | StorageFailure
    > =>
      Effect.gen(function* () {
        yield* guardOrgWrite(input.owner);
        const name = connectionIdentifier(String(input.name));
        // Typed (not StorageError) so the HTTP edge can answer 400 with the
        // reason instead of an opaque 500 — callers can act on it.
        if (input.owner === "user" && subject == null) {
          return yield* new InvalidConnectionInputError({
            message:
              'Cannot create a personal connection: this context has no user subject. Create it with owner "org", or connect as a signed-in user.',
          });
        }
        const integrationRow = yield* findIntegrationRow(input.integration);
        if (!integrationRow) {
          return yield* new IntegrationNotFoundError({
            slug: input.integration,
          });
        }

        // Create is never a replace. This early check answers the common case
        // with a typed 409 before any other work, but it is NOT the guard
        // against concurrent creates — the row insert below is: the
        // transaction re-checks, the primary key breaks the tie, and the
        // provider write happens only after the insert wins.
        const duplicate = yield* findConnectionRow({
          owner: input.owner,
          integration: input.integration,
          name,
        });
        let retryingRowId: string | null = null;
        let retryingItemIds: readonly string[] = [];
        if (duplicate) {
          const duplicateItemIds = Object.values(connectionItemIds(duplicate));
          const duplicateProvider = credentialProviders.get(duplicate.provider);
          const duplicateAttempt = parseCredentialWriteAttempt(duplicate.credential_write);
          const belongsToCrashedRuntime =
            duplicateItemIds.length > 0 &&
            duplicateAttempt !== null &&
            duplicateAttempt.runtimeId !== credentialWriteRuntimeId;
          const hasMissingCredential =
            belongsToCrashedRuntime && duplicateProvider?.set
              ? (yield* Effect.forEach(duplicateItemIds, (itemId) =>
                  duplicateProvider
                    .get(ProviderItemId.make(itemId))
                    .pipe(Effect.map((value) => value === null)),
                )).some(Boolean)
              : false;
          retryingRowId = hasMissingCredential ? storageRowId(duplicate) : null;