Introduce waiters for Intake Runner, Intake and Intake User

CHANGELOG.md

@@ -1,9 +1,13 @@
+
 ## Release (2025-xx-xx)
 - `stackitmarketplace`: [v1.15.0](services/stackitmarketplace/CHANGELOG.md#v1150)
   - **Feature:** Add `EndUserLicenseAgreement`, `ProductDescription` and `ServiceLevelAgreement` attributes and add them to `Assets` struct
 - `postgresflex`: [v1.3.0](services/postgresflex/CHANGELOG.md#v130)
   - **Breaking Change:** The attribute type for `PartialUpdateInstancePayload` and `UpdateInstancePayload` changed from `Storage` to `StorageUpdate`.
   - **Deprecation:** `StorageUpdate`: updating the performance class field is not possible.
+- `intake`: [v0.3.0](services/intake/CHANGELOG.md#v030)
+  - **Feature:** Add wait handlers for `Intake`, `IntakeRunner`, and `IntakeUser` resources.
+  - **Improvement:** Add usage examples for the `intake` service.
 
 ## Release (2025-10-13)
 - `observability`: [v0.15.0](services/observability/CHANGELOG.md#v0150)

examples/intake/go.mod

@@ -0,0 +1,13 @@
+module github.com/stackitcloud/stackit-sdk-go/examples/intake
+
+go 1.21
+
+require (
+	github.com/stackitcloud/stackit-sdk-go/core v0.17.3
+	github.com/stackitcloud/stackit-sdk-go/services/intake v0.2.0
+)
+
+require (
+	github.com/golang-jwt/jwt/v5 v5.3.0 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+)

examples/intake/go.sum

@@ -0,0 +1,9 @@
+github.com/golang-jwt/jwt/v5 v5.3.0 h1:pv4AsKCKKZuqlgs5sUmn4x8UlGa0kEVt/puTpKx9vvo=
+github.com/golang-jwt/jwt/v5 v5.3.0/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/stackitcloud/stackit-sdk-go/core v0.17.3 h1:GsZGmRRc/3GJLmCUnsZswirr5wfLRrwavbnL/renOqg=
+github.com/stackitcloud/stackit-sdk-go/core v0.17.3/go.mod h1:HBCXJGPgdRulplDzhrmwC+Dak9B/x0nzNtmOpu+1Ahg=
+github.com/stackitcloud/stackit-sdk-go/services/intake v0.2.0 h1:p/zi4VPoCQWk7/2ubi3hxsqiaye41x/Pl3GXYbPkYOY=
+github.com/stackitcloud/stackit-sdk-go/services/intake v0.2.0/go.mod h1:jOArPjNRkwv4487+9ab3dRG+lM09leu5FiRohbQs9Z4=

examples/intake/intake.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	sdkConfig "github.com/stackitcloud/stackit-sdk-go/core/config"
+	"github.com/stackitcloud/stackit-sdk-go/core/utils"
+	"github.com/stackitcloud/stackit-sdk-go/services/intake"
+	"github.com/stackitcloud/stackit-sdk-go/services/intake/wait"
+)
+
+func main() {
+	region := "eu01"          // Region where the resources will be created
+	projectId := "PROJECT_ID" // Your STACKIT project ID
+
+	dremioCatalogURI := "DREMIO_CATALOG_URI"       // E.g., "https://my-dremio-catalog.data-platform.stackit.run/iceberg"
+	dremioTokenEndpoint := "DREMIO_TOKEN_ENDPOINT" // E.g., "https://my-dremio.data-platform.stackit.run/oauth/token"
+	dremioPAT := "DREMIO_PERSONAL_ACCESS_TOKEN"    // Your Dremio Personal Access Token
+	catalogWarehouse := "CATALOG_WAREHOUSE"        // Catalog warehouse where the data will be ingested
+
+	intakeUserPassword := "s3cuRe_p@ssW0rd_f0r_1ntake!" // A secure password for the new intake user
+
+	ctx := context.Background()
+
+	intakeClient, err := intake.NewAPIClient(
+		sdkConfig.WithRegion(region),
+	)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Creating API client: %v\n", err)
+		os.Exit(1)
+	}
+
+	// Create an Intake Runner
+	createRunnerPayload := intake.CreateIntakeRunnerPayload{
+		DisplayName:        utils.Ptr("my-example-runner"),
+		MaxMessageSizeKiB:  utils.Ptr(int64(10)),
+		MaxMessagesPerHour: utils.Ptr(int64(1000)),
+	}
+	createRunnerResp, err := intakeClient.CreateIntakeRunner(ctx, projectId, region).CreateIntakeRunnerPayload(createRunnerPayload).Execute()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating Intake Runner: %v\n", err)
+		os.Exit(1)
+	}
+	intakeRunnerId := *createRunnerResp.Id
+	fmt.Printf("Triggered creation of Intake Runner with ID: %s. Waiting for it to become active...\n", intakeRunnerId)
+
+	// Wait for the Intake Runner to become active
+	_, err = wait.CreateOrUpdateIntakeRunnerWaitHandler(ctx, intakeClient, projectId, region, intakeRunnerId).WaitWithContext(ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error waiting for Intake Runner: %v\n", err)
+		os.Exit(1)
+	}
+
+	// Create an Intake
+	dremioAuthType := intake.CatalogAuthType("dremio") // can also be set to "none" if the catalog is not authenticated
+	createIntakePayload := intake.CreateIntakePayload{
+		DisplayName:    utils.Ptr("my-example-intake"),
+		IntakeRunnerId: utils.Ptr(intakeRunnerId),
+		Catalog: &intake.IntakeCatalog{
+			Uri:       utils.Ptr(dremioCatalogURI),
+			Warehouse: utils.Ptr(catalogWarehouse),
+			Namespace: utils.Ptr("example_namespace"),
+			TableName: utils.Ptr("example_table"),
+			Auth: &intake.CatalogAuth{
+				Type: &dremioAuthType,
+				Dremio: &intake.DremioAuth{
+					TokenEndpoint:       utils.Ptr(dremioTokenEndpoint),
+					PersonalAccessToken: utils.Ptr(dremioPAT),
+				},
+			},
+		},
+	}
+	createIntakeResp, err := intakeClient.CreateIntake(ctx, projectId, region).CreateIntakePayload(createIntakePayload).Execute()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating Intake: %v\n", err)
+		os.Exit(1)
+	}
+	intakeId := *createIntakeResp.Id
+	fmt.Printf("Triggered creation of Intake with ID: %s. Waiting for it to become active...\n", intakeRunnerId)
+
+	// Wait for the Intake to become active
+	_, err = wait.CreateOrUpdateIntakeWaitHandler(ctx, intakeClient, projectId, region, intakeId).WaitWithContext(ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error waiting for Intake: %v\n", err)
+		os.Exit(1)
+	}
+
+	createIntakeUserPayload := intake.CreateIntakeUserPayload{
+		DisplayName: utils.Ptr("my-example-user"),
+		Password:    utils.Ptr(intakeUserPassword),
+	}
+	createIntakeUserResp, err := intakeClient.CreateIntakeUser(ctx, projectId, region, intakeId).CreateIntakeUserPayload(createIntakeUserPayload).Execute()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating Intake User: %v\n", err)
+		os.Exit(1)
+	}
+	intakeUserId := *createIntakeUserResp.Id
+	fmt.Printf("Triggered creation of Intake User with ID: %s. Waiting for it to become active...\n", intakeRunnerId)
+
+	// Wait for the Intake User to become active
+	_, err = wait.CreateOrUpdateIntakeUserWaitHandler(ctx, intakeClient, projectId, region, intakeId, intakeUserId).WaitWithContext(ctx)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error waiting for Intake User: %v\n", err)
+		os.Exit(1)
+	}
+}

go.work

@@ -10,6 +10,7 @@ use (
 	./examples/dns
 	./examples/errorhandling
 	./examples/iaas
+	./examples/intake
 	./examples/kms
 	./examples/loadbalancer
 	./examples/logme

services/intake/CHANGELOG.md

@@ -1,3 +1,7 @@
+## v0.3.0
+- **Feature:** Add wait handlers for `Intake`, `IntakeRunner`, and `IntakeUser` resources.
+- **Improvement:** Add usage examples for the `intake` service.
+
 ## v0.2.0
 - **Feature:** Add response `IntakeRunnerResponse` to `UpdateIntakeRunnerExecute` request
 - **Feature:** Add response `IntakeUserResponse` to `UpdateIntakeUserExecute` request

services/intake/VERSION

@@ -1 +1 @@
-v0.2.0
+v0.3.0

services/intake/go.mod

@@ -3,6 +3,7 @@ module github.com/stackitcloud/stackit-sdk-go/services/intake
 go 1.21
 
 require (
+	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/stackitcloud/stackit-sdk-go/core v0.17.3
 )

services/intake/wait/wait.go

@@ -0,0 +1,162 @@
+package wait
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/stackitcloud/stackit-sdk-go/core/oapierror"
+	"github.com/stackitcloud/stackit-sdk-go/core/wait"
+	"github.com/stackitcloud/stackit-sdk-go/services/intake"
+)
+
+type APIClientInterface interface {
+	GetIntakeRunnerExecute(ctx context.Context, projectId, region, intakeRunnerId string) (*intake.IntakeRunnerResponse, error)
+	GetIntakeExecute(ctx context.Context, projectId, region, intakeId string) (*intake.IntakeResponse, error)
+	GetIntakeUserExecute(ctx context.Context, projectId, region, intakeId, intakeUserId string) (*intake.IntakeUserResponse, error)
+}
+
+func CreateOrUpdateIntakeRunnerWaitHandler(ctx context.Context, a APIClientInterface, projectId, region, intakeRunnerId string) *wait.AsyncActionHandler[intake.IntakeRunnerResponse] {
+	handler := wait.New(func() (waitFinished bool, response *intake.IntakeRunnerResponse, err error) {
+		runner, err := a.GetIntakeRunnerExecute(ctx, projectId, region, intakeRunnerId)
+		if err != nil {
+			return false, nil, err
+		}
+
+		if runner == nil {
+			return false, nil, fmt.Errorf("API returned a nil response for Intake Runner %s", intakeRunnerId)
+		}
+
+		if runner.Id == nil || runner.State == nil {
+			return false, nil, fmt.Errorf("could not get ID or State from response for Intake Runner %s", intakeRunnerId)
+		}
+
+		if *runner.Id == intakeRunnerId && *runner.State == intake.INTAKERUNNERRESPONSESTATE_ACTIVE {
+			return true, runner, nil
+		}
+
+		// The API does not have a dedicated failure state for this resource,
+		// so we rely on the timeout for cases where it never becomes active.
+		return false, nil, nil
+	})
+	handler.SetTimeout(15 * time.Minute)
+	return handler
+}
+
+func DeleteIntakeRunnerWaitHandler(ctx context.Context, a APIClientInterface, projectId, region, intakeRunnerId string) *wait.AsyncActionHandler[intake.IntakeRunnerResponse] {
+	handler := wait.New(func() (waitFinished bool, response *intake.IntakeRunnerResponse, err error) {
+		_, err = a.GetIntakeRunnerExecute(ctx, projectId, region, intakeRunnerId)
+		if err == nil {
+			// Resource still exists
+			return false, nil, nil
+		}
+
+		var oapiError *oapierror.GenericOpenAPIError
+		if errors.As(err, &oapiError) {
+			if oapiError.StatusCode == http.StatusNotFound {
+				// Success: Resource is gone
+				return true, nil, nil
+			}
+		}
+		// An unexpected error occurred
+		return false, nil, err
+	})
+	handler.SetTimeout(15 * time.Minute)
+	return handler
+}
+
+func CreateOrUpdateIntakeWaitHandler(ctx context.Context, a APIClientInterface, projectId, region, intakeId string) *wait.AsyncActionHandler[intake.IntakeResponse] {
+	handler := wait.New(func() (waitFinished bool, response *intake.IntakeResponse, err error) {
+		ik, err := a.GetIntakeExecute(ctx, projectId, region, intakeId)
+		if err != nil {
+			return false, nil, err
+		}
+
+		if ik == nil {
+			return false, nil, fmt.Errorf("API returned a nil response for Intake %s", intakeId)
+		}
+
+		if ik.Id == nil || ik.State == nil {
+			return false, nil, fmt.Errorf("could not get ID or State from response for Intake %s", intakeId)
+		}
+
+		if *ik.Id == intakeId && *ik.State == intake.INTAKERESPONSESTATE_ACTIVE {
+			return true, ik, nil
+		}
+
+		if *ik.Id == intakeId && *ik.State == intake.INTAKERESPONSESTATE_FAILED {
+			return true, ik, fmt.Errorf("create/update failed for Intake %s", intakeId)
+		}
+
+		return false, nil, nil
+	})
+	handler.SetTimeout(10 * time.Minute)
+	return handler
+}
+
+func DeleteIntakeWaitHandler(ctx context.Context, a APIClientInterface, projectId, region, intakeId string) *wait.AsyncActionHandler[intake.IntakeResponse] {
+	handler := wait.New(func() (waitFinished bool, response *intake.IntakeResponse, err error) {
+		_, err = a.GetIntakeExecute(ctx, projectId, region, intakeId)
+		if err == nil {
+			return false, nil, nil
+		}
+
+		var oapiError *oapierror.GenericOpenAPIError
+		if errors.As(err, &oapiError) {
+			if oapiError.StatusCode == http.StatusNotFound {
+				return true, nil, nil
+			}
+		}
+		return false, nil, err
+	})
+	handler.SetTimeout(10 * time.Minute)
+	return handler
+}
+
+func CreateOrUpdateIntakeUserWaitHandler(ctx context.Context, a APIClientInterface, projectId, region, intakeId, intakeUserId string) *wait.AsyncActionHandler[intake.IntakeUserResponse] {
+	handler := wait.New(func() (waitFinished bool, response *intake.IntakeUserResponse, err error) {
+		user, err := a.GetIntakeUserExecute(ctx, projectId, region, intakeId, intakeUserId)
+		if err != nil {
+			return false, nil, err
+		}
+
+		if user == nil {
+			return false, nil, fmt.Errorf("API returned a nil response for Intake User %s", intakeUserId)
+		}
+
+		if user.Id == nil || user.State == nil {
+			return false, nil, fmt.Errorf("could not get ID or State from response for Intake User %s", intakeUserId)
+		}
+
+		if *user.Id == intakeUserId && *user.State == intake.INTAKEUSERRESPONSESTATE_ACTIVE {
+			return true, user, nil
+		}
+
+		// The API does not have a dedicated failure state for this resource, we rely on the timeout for cases where
+		// it never becomes active.
+		return false, nil, nil
+	})
+	handler.SetTimeout(5 * time.Minute)
+	return handler
+}
+
+func DeleteIntakeUserWaitHandler(ctx context.Context, a APIClientInterface, projectId, region, intakeId, intakeUserId string) *wait.AsyncActionHandler[intake.IntakeUserResponse] {
+	handler := wait.New(func() (waitFinished bool, response *intake.IntakeUserResponse, err error) {
+		_, err = a.GetIntakeUserExecute(ctx, projectId, region, intakeId, intakeUserId)
+		if err == nil {
+			return false, nil, nil
+		}
+
+		var oapiError *oapierror.GenericOpenAPIError
+		if errors.As(err, &oapiError) {
+			if oapiError.StatusCode == http.StatusNotFound {
+				return true, nil, nil
+			}
+		}
+		return false, nil, err
+	})
+	handler.SetTimeout(5 * time.Minute)
+	return handler
+}