direct: pass last-persisted state to DoDelete (#5279)

## Summary

Extends `IResource.DoDelete` to take the resource's last-persisted state
alongside `id`. The framework loads it from `dstate` in
`DeploymentUnit.Delete` and `DeploymentUnit.Recreate`, and
`Adapter.DoDelete` forwards it to the resource implementation. No
behavioral change in this PR: every existing resource ignores the new
argument (`_ *FooState`).

Motivation: some delete APIs need values that come from local config
(e.g. `postgres_project.purge_on_delete`, which is set on the user's
resource and forwarded to the Lakebase DeleteProject RPC). Today there
is nowhere to plumb those values because `DoDelete` only sees the
resource id. This change keeps the call symmetric with
`DoCreate`/`DoUpdate`, which already receive state.

Adapter-level validation now type-checks `DoDelete state` the same way
it already does for `DoCreate newState`.

## Test plan

- [x] Unit + acceptance suites pass (`go test ./bundle/...`, `go test
./acceptance/...`); no fixture updates needed since behavior is
unchanged.

This pull request and its description were written by Isaac.

Author: pietern

bundle/direct/apply.go

@@ -80,11 +80,16 @@ func (d *DeploymentUnit) Create(ctx context.Context, db *dstate.DeploymentState,
 }
 
 func (d *DeploymentUnit) Recreate(ctx context.Context, db *dstate.DeploymentState, oldID string, newState any) error {
+	oldState, err := d.loadPersistedState(db)
+	if err != nil {
+		return err
+	}
+
 	// Note, unlike Delete(), we hard error on 403 here intentionally.
 	// MANAGED_BY_PARENT is still disregarded — the subsequent Create with
 	// replace_existing=true will reconfigure the parent-managed resource in
 	// place, matching the Terraform provider's recreate behaviour.
-	err := d.Adapter.DoDelete(ctx, oldID)
+	err = d.Adapter.DoDelete(ctx, oldID, oldState)
 	if err != nil && !isResourceGone(err) && !isManagedByParent(err) {
 		return fmt.Errorf("deleting old id=%s: %w", oldID, err)
 	}
@@ -170,7 +175,12 @@ func (d *DeploymentUnit) UpdateWithID(ctx context.Context, db *dstate.Deployment
 }
 
 func (d *DeploymentUnit) Delete(ctx context.Context, db *dstate.DeploymentState, oldID string) error {
-	err := d.Adapter.DoDelete(ctx, oldID)
+	oldState, err := d.loadPersistedState(db)
+	if err != nil {
+		return err
+	}
+
+	err = d.Adapter.DoDelete(ctx, oldID, oldState)
 	if err != nil && !isResourceGone(err) && !isManagedByParent(err) {
 		// Rather than failing delete and requiring user to unbind, we perform unbind automatically there.
 		// Some services, e.g. jobs, return 403 for missing resources if caller did not have permissions to it when job existed.
@@ -215,6 +225,23 @@ func parseState(destType reflect.Type, raw json.RawMessage) (any, error) {
 	return reflect.ValueOf(destPtr).Elem().Interface(), nil
 }
 
+// loadPersistedState reads and parses the resource's last-persisted state for
+// the DoDelete call. Returns a zero-value state pointer when nothing has been
+// persisted yet (e.g. delete after a partial-create failure), so the call site
+// always passes a typed value.
+func (d *DeploymentUnit) loadPersistedState(db *dstate.DeploymentState) (any, error) {
+	stateType := d.Adapter.StateType()
+	dbentry, ok := db.GetResourceEntry(d.ResourceKey)
+	if !ok || len(dbentry.State) == 0 {
+		return reflect.New(stateType.Elem()).Interface(), nil
+	}
+	state, err := parseState(stateType, dbentry.State)
+	if err != nil {
+		return nil, fmt.Errorf("parsing persisted state: %w", err)
+	}
+	return state, nil
+}
+
 func (d *DeploymentUnit) refreshRemoteState(ctx context.Context, id string) error {
 	if d.RemoteState != nil {
 		return nil

bundle/direct/dresources/adapter.go

@@ -42,9 +42,11 @@ type IResource interface {
 	// Example: func (r *ResourceJob) DoRead(ctx context.Context, id string) (*jobs.Job, error)
 	DoRead(ctx context.Context, id string) (remoteState any, e error)
 
-	// DoDelete deletes the resource.
-	// Example: func (r *ResourceJob) DoDelete(ctx context.Context, id string) error
-	DoDelete(ctx context.Context, id string) error
+	// DoDelete deletes the resource. The state argument is the last-persisted
+	// state for the resource; resources that don't need it should accept it as
+	// _ to satisfy the interface.
+	// Example: func (r *ResourceJob) DoDelete(ctx context.Context, id string, _ *jobs.JobSettings) error
+	DoDelete(ctx context.Context, id string, state any) error
 
 	// [Optional] OverrideChangeDesc can implement custom logic to update a given ChangeDesc; it is run last after built-in classifiers and field triggers.
 	OverrideChangeDesc(ctx context.Context, path *structpath.PathNode, changedesc *ChangeDesc, remoteState any) error
@@ -264,6 +266,7 @@ func (a *Adapter) validate() error {
 	validations := []any{
 		"PrepareState return", a.prepareState.OutTypes[0], stateType,
 		"DoCreate newState", a.doCreate.InTypes[1], stateType,
+		"DoDelete state", a.doDelete.InTypes[2], stateType,
 	}
 
 	// If RemapState is implemented, validate its signature.
@@ -399,12 +402,9 @@ func (a *Adapter) DoRead(ctx context.Context, id string) (any, error) {
 	return outs[0], nil
 }
 
-func (a *Adapter) DoDelete(ctx context.Context, id string) error {
-	_, err := a.doDelete.Call(ctx, id)
-	if err != nil {
-		return err
-	}
-	return nil
+func (a *Adapter) DoDelete(ctx context.Context, id string, state any) error {
+	_, err := a.doDelete.Call(ctx, id, state)
+	return err
 }
 
 // normalizeNilPointer converts a nil pointer wrapped in an interface to a nil interface.

bundle/direct/dresources/alert.go

@@ -59,7 +59,7 @@ func (r *ResourceAlert) DoUpdate(ctx context.Context, id string, config *sql.Ale
 }
 
 // DoDelete deletes the alert by id.
-func (r *ResourceAlert) DoDelete(ctx context.Context, id string) error {
+func (r *ResourceAlert) DoDelete(ctx context.Context, id string, _ *sql.AlertV2) error {
 	return r.client.AlertsV2.TrashAlert(ctx, sql.TrashAlertV2Request{
 		Id:              id,
 		Purge:           true,

bundle/direct/dresources/all_test.go

@@ -860,7 +860,7 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 		assert.Equal(t, val, remoteValue, "path=%q\nnewState=%s\nremappedState=%s", path.String(), jsonDump(newState), jsonDump(remappedState))
 	}))
 
-	err = adapter.DoDelete(ctx, createdID)
+	err = adapter.DoDelete(ctx, createdID, newState)
 	require.NoError(t, err)
 
 	p, err := structpath.ParsePath("name")

bundle/direct/dresources/app.go

@@ -316,7 +316,7 @@ func deploymentToAppConfig(d *apps.AppDeployment) *resources.AppConfig {
 	return config
 }
 
-func (r *ResourceApp) DoDelete(ctx context.Context, id string) error {
+func (r *ResourceApp) DoDelete(ctx context.Context, id string, _ *AppState) error {
 	_, err := r.client.Apps.DeleteByName(ctx, id)
 	return err
 }

bundle/direct/dresources/catalog.go

@@ -104,7 +104,7 @@ func (r *ResourceCatalog) DoUpdateWithID(ctx context.Context, id string, config
 	return newID, response, nil
 }
 
-func (r *ResourceCatalog) DoDelete(ctx context.Context, id string) error {
+func (r *ResourceCatalog) DoDelete(ctx context.Context, id string, _ *catalog.CreateCatalog) error {
 	return r.client.Catalogs.Delete(ctx, catalog.DeleteCatalogRequest{
 		Name:            id,
 		Force:           true,

bundle/direct/dresources/cluster.go

@@ -117,7 +117,7 @@ func (r *ResourceCluster) DoResize(ctx context.Context, id string, config *compu
 	return err
 }
 
-func (r *ResourceCluster) DoDelete(ctx context.Context, id string) error {
+func (r *ResourceCluster) DoDelete(ctx context.Context, id string, _ *compute.ClusterSpec) error {
 	return r.client.Clusters.PermanentDeleteByClusterId(ctx, id)
 }
 

bundle/direct/dresources/dashboard.go

@@ -357,7 +357,7 @@ func (r *ResourceDashboard) DoUpdate(ctx context.Context, id string, config *Das
 	return responseToState(updateResp, publishResp, dashboard.SerializedDashboard, config.Published), nil
 }
 
-func (r *ResourceDashboard) DoDelete(ctx context.Context, id string) error {
+func (r *ResourceDashboard) DoDelete(ctx context.Context, id string, _ *DashboardState) error {
 	return r.client.Lakeview.Trash(ctx, dashboards.TrashDashboardRequest{
 		DashboardId: id,
 	})

bundle/direct/dresources/database_catalog.go

@@ -45,7 +45,7 @@ func (r *ResourceDatabaseCatalog) DoUpdate(ctx context.Context, id string, confi
 	return nil, err
 }
 
-func (r *ResourceDatabaseCatalog) DoDelete(ctx context.Context, id string) error {
+func (r *ResourceDatabaseCatalog) DoDelete(ctx context.Context, id string, _ *database.DatabaseCatalog) error {
 	return r.client.Database.DeleteDatabaseCatalog(ctx, database.DeleteDatabaseCatalogRequest{
 		Name: id,
 	})

bundle/direct/dresources/database_instance.go

@@ -60,7 +60,7 @@ func (d *ResourceDatabaseInstance) WaitAfterCreate(ctx context.Context, id strin
 	return nil, err
 }
 
-func (d *ResourceDatabaseInstance) DoDelete(ctx context.Context, name string) error {
+func (d *ResourceDatabaseInstance) DoDelete(ctx context.Context, name string, _ *database.DatabaseInstance) error {
 	return d.client.Database.DeleteDatabaseInstance(ctx, database.DeleteDatabaseInstanceRequest{
 		Name:            name,
 		Purge:           true,