package server

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/tenant"
	"go.uber.org/zap"
)

const (
	defaultTenantDeletePollInterval = time.Minute
	defaultTenantDeleteRetryDelay   = time.Hour
	defaultTenantDeleteRunningTTL   = 30 * time.Minute
	defaultTenantDeleteJobTimeout   = 25 * time.Minute
	// defaultTenantDeleteJobFallbackDelay is crash-fallback headroom for
	// resumable shared-schema purges. Cluster-backed deletes create their job
	// only after cluster deletion succeeds.
	defaultTenantDeleteJobFallbackDelay = 10 * time.Minute
)

// detachedDeleteWorkContext returns a context that survives client
// disconnects (so an interrupted request cannot abort a delete halfway) but
// is still cancelled on server shutdown when the fork worker context is
// available.
func (s *Server) detachedDeleteWorkContext(ctx context.Context) context.Context {
	if s.forkWorkerCtx != nil {
		return contextWithTrace(s.forkWorkerCtx, ctx)
	}
	return backgroundWithTrace(ctx)
}

// drainTenantBackend terminates the tenant's established SSE streams (so
// their in-flight requests release the pool backend reference pinned by
// auth) and then waits for the backend references to drain.
//
// A /v1/events connect that passed auth just before the deleting mark can
// resolve its bus AFTER the first closeBus and re-create a never-closed one,
// re-pinning a reference until the client disconnects. On drain timeout,
// close once more and re-drain: by then the tenant is already deleting, so
// the auth gate rejects new connects before they can Acquire and the second
// pass cannot lose the same race. The re-drain itself returns promptly —
// the first InvalidateAndWait already retired the cache entry, so the
// second call only confirms no new entry appeared. handleEvents also
// re-checks tenant status before subscribing, which shrinks the race window
// to begin with.
func (s *Server) drainTenantBackend(ctx context.Context, tenantID string) error {
	s.events.closeBus(ctx, tenantID)
	err := s.pool.InvalidateAndWait(ctx, tenantID)
	if err == nil {
		return nil
	}
	if !errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {
		// Real cancellation (e.g. shutdown) or a non-timeout failure:
		// propagate instead of proceeding with a live backend.
		return err
	}
	s.events.closeBus(ctx, tenantID)
	return s.pool.InvalidateAndWait(ctx, tenantID)
}

func (s *Server) handleTenantDelete(w http.ResponseWriter, r *http.Request) {
	if s.meta == nil || s.pool == nil || (s.provisioner == nil && s.legacyStarterProvisioner == nil) || len(s.tokenSecret) == 0 {
		errJSON(w, http.StatusNotFound, "tenant delete not enabled")
		return
	}
	if r.Method != http.MethodDelete {
		errJSON(w, http.StatusMethodNotAllowed, "method not allowed")
		return
	}
	scope, ok := ownerScopeFromRequest(w, r, "delete tenant")
	if !ok {
		return
	}
	if scope == nil || scope.TenantID == "" {
		errJSON(w, http.StatusUnauthorized, "missing tenant scope")
		return
	}

	t, err := s.meta.GetTenant(r.Context(), scope.TenantID)
	if err != nil {
		if errors.Is(err, meta.ErrNotFound) {
			errJSON(w, http.StatusNotFound, "tenant not found")
			return
		}
		errJSON(w, backendErrorStatus(r.Context(), err), "tenant lookup failed")
		return
	}
	if t.Kind != meta.TenantKindLive {
		errJSON(w, http.StatusConflict, "only live tenants can be deleted")
		return
	}
	if t.Status == meta.TenantDeleted {
		errJSON(w, http.StatusNotFound, "tenant not found")
		return
	}
	// Shared-schema tenants follow a dedicated delete path: purge their rows
	// from the shared DB instead of deprovisioning a cluster (there is none).
	// The dispatch keys on the persisted provider, not the placement row —
	// the provider never changes mid-delete, so a retried delete always
	// routes back here even after the placement row has been removed.
	if tenant.IsSharedSchemaProvider(t.Provider) {
		s.handleSharedTenantDelete(w, r, t)
		return
	}
	if t.Provider == tenant.ProviderTiDBZero {
		errJSON(w, http.StatusConflict, "tidb_zero tenants expire automatically and do not support delete")
		return
	}
	if !tenant.SupportsClusterDelete(t.Provider) {
		errJSON(w, http.StatusConflict, "tenant delete is not supported for provider")
		return
	}
	if t.StorageNamespaceID != "" {
		hasFork, err := s.meta.NamespaceHasNonDeletedFork(r.Context(), t.StorageNamespaceID)
		if err != nil {
			errJSON(w, backendErrorStatus(r.Context(), err), "failed to check tenant forks")
			return
		}
		if hasFork {
			errJSON(w, http.StatusConflict, "tenant has non-deleted forks")
			return
		}
	}

	var req tenant.CredentialProvisionRequest
	if t.Status == meta.TenantDeleting {
		armed, err := s.meta.TenantDeleteJobArmed(r.Context(), t.ID)
		if err != nil {
			errJSON(w, backendErrorStatus(r.Context(), err), "failed to check tenant delete cleanup")
			return
		}
		if armed {
			s.clearLocalTenantMetrics(t.ID)
			_ = s.meta.RevokeTenantAPIKeys(r.Context(), t.ID)
			writeTenantDeleteStatus(w, meta.TenantDeleting)
			return
		}
	}

	if t.Provider == tenant.ProviderTiDBCloudNative {
		req, err = decodeCredentialDeprovisionRequest(w, r)
		if err != nil {
			if !errors.Is(err, tenant.ErrCredentialsRequired) {
				errJSON(w, http.StatusBadRequest, err.Error())
				return
			}
			defaultReq := resolveDefaultCredentials(s.provisioner)
			if defaultReq == nil {
				errJSON(w, http.StatusBadRequest, tenant.ErrCredentialsRequired.Error())
				return
			}
			req = *defaultReq
		}
	}

	// The deleting mark is durable before external deprovision starts. The
	// cleanup job is created only after cluster deletion succeeds, so elapsed
	// time can never start object cleanup first. The detached context survives
	// client disconnects; a later request can re-drive an interrupted cluster
	// delete while the tenant remains deleting without a job.
	workCtx := s.detachedDeleteWorkContext(r.Context())
	updated, err := s.markTenantDeletingForCleanup(workCtx, t)
	if err != nil {
		errJSON(w, backendErrorStatus(r.Context(), err), "failed to mark tenant deleting")
		return
	}
	if !updated {
		writeTenantDeleteStatus(w, meta.TenantDeleting)
		return
	}
	if err := s.drainTenantBackend(workCtx, t.ID); err != nil {
		errJSON(w, backendErrorStatus(r.Context(), err), "failed to drain tenant backend")
		return
	}
	if err := s.deprovisionTenantCluster(workCtx, t, req); err != nil {
		status, msg := clientFacingErrorResponse(http.StatusBadGateway, "delete tenant cluster failed", err)
		errJSON(w, status, msg)
		return
	}

	_ = s.meta.AbortActiveUploadReservations(workCtx, t.ID)

	status, err := s.enqueueTenantDeleteJob(workCtx, t)
	if err != nil {
		// Cluster deprovision already succeeded. Keep owner keys active until
		// cleanup is durably enqueued so the same owner can retry the delete
		// request. This matters for tidb_cloud_native because customer TiDB Cloud
		// credentials are accepted per request and are not stored server-side.
		errJSON(w, backendErrorStatus(r.Context(), err), "failed to enqueue tenant delete cleanup")
		return
	}
	_ = s.meta.RevokeTenantAPIKeys(workCtx, t.ID)
	writeTenantDeleteStatus(w, status)
}

// rollbackTenantDeleteMark undoes a shared-schema deleting mark when purge
// fails before destructive work starts, dropping its delayed cleanup job in
// the same transaction. A no-op rollback (tenant already progressed by
// another actor) is logged, since the caller reports the purge failure either
// way. Cluster-backed deletes never call this after their job is created.
func (s *Server) rollbackTenantDeleteMark(ctx context.Context, tenantID, namespaceID string, to meta.TenantStatus) {
	rollbackCtx := backgroundWithTrace(ctx)
	if namespaceID != "" {
		rolledBack, err := s.meta.RollbackTenantDeletingWithDeleteJob(rollbackCtx, tenantID, namespaceID, to)
		if err != nil {
			logger.Warn(rollbackCtx, "tenant_delete_rollback_failed",
				zap.String("tenant_id", tenantID), zap.Error(err))
			return
		}
		if !rolledBack {
			logger.Warn(rollbackCtx, "tenant_delete_rollback_skipped",
				zap.String("tenant_id", tenantID),
				zap.String("reason", "delete no longer rollbackable"))
		}
		return
	}
	_, _ = s.meta.UpdateTenantStatusIf(rollbackCtx, tenantID, meta.TenantDeleting, to)
}

// handleSharedTenantDelete deletes a tenant placed on a shared-schema
// database: it purges the tenant's rows from the shared DB (never touching
// other tenants' rows and never deprovisioning the shared cluster), then runs
// the same S3-namespace cleanup and key revocation as the standalone path.
//
// The flow is safe to retry at any point: the dispatch keys on the persisted
// provider (durable), the purge is idempotent, and external cleanup retains a
// deleting placement as the physical-resource authorization anchor until the
// cleanup worker reaches the terminal state. Identity resolution is read-only
// (ResolveFsID): a delete path must never allocate a new fs_id.
func (s *Server) handleSharedTenantDelete(w http.ResponseWriter, r *http.Request, t *meta.Tenant) {
	s.handleSharedTenantDeleteWithStatusWriter(w, r, t, writeTenantDeleteStatus)
}

func (s *Server) handleSharedTenantDeleteWithStatusWriter(
	w http.ResponseWriter,
	r *http.Request,
	t *meta.Tenant,
	writeStatus func(http.ResponseWriter, meta.TenantStatus),
) {
	ctx := s.detachedDeleteWorkContext(r.Context())
	deleteJobArmed := false
	if t.StorageNamespaceID != "" {
		hasFork, err := s.meta.NamespaceHasNonDeletedFork(ctx, t.StorageNamespaceID)
		if err != nil {
			errJSON(w, backendErrorStatus(r.Context(), err), "failed to check tenant forks")
			return
		}
		if hasFork {
			errJSON(w, http.StatusConflict, "tenant has non-deleted forks")
			return
		}
	}
	if t.Status == meta.TenantDeleting {
		armed, err := s.meta.TenantDeleteJobArmed(ctx, t.ID)
		if err != nil {
			errJSON(w, backendErrorStatus(r.Context(), err), "failed to check tenant delete cleanup")
			return
		}
		deleteJobArmed = armed
	}
	if t.Status != meta.TenantDeleting || (t.StorageNamespaceID != "" && !deleteJobArmed) {
		// The deleting mark and delayed cleanup job land in one transaction;
		// the trailing enqueueTenantDeleteJob call makes the job immediately due
		// once the purge completes.
		updated, err := s.markTenantDeletingForCleanup(ctx, t)
		if err != nil {
			errJSON(w, backendErrorStatus(r.Context(), err), "failed to mark tenant deleting")
			return
		}
		if !updated {
			writeStatus(w, meta.TenantDeleting)
			return
		}
	}
	markedDeleting := t.Status != meta.TenantDeleting
	if err := s.drainTenantBackend(ctx, t.ID); err != nil {
		if markedDeleting {
			s.rollbackTenantDeleteMark(ctx, t.ID, t.StorageNamespaceID, t.Status)
		}
		errJSON(w, backendErrorStatus(r.Context(), err), "failed to drain tenant backend")
		return
	}
	fsID, err := s.meta.ResolveFsID(ctx, t.ID)
	if err != nil {
		// A shared-schema tenant without an fs_registry row is a data
		// integrity problem; never allocate one on a delete path.
		logger.Error(ctx, "shared_tenant_delete_fsid_unresolvable", zap.String("tenant_id", t.ID), zap.Error(err))
		errJSON(w, backendErrorStatus(r.Context(), err), "failed to resolve tenant fs_id")
		return
	}
	placement, err := s.meta.GetTenantPlacement(ctx, fsID)
	if err != nil && !errors.Is(err, meta.ErrNotFound) {
		errJSON(w, backendErrorStatus(r.Context(), err), "failed to resolve tenant placement")
		return
	}
	if deleteJobArmed && placement != nil && placement.Status == meta.PlacementStatusDeleting {
		s.clearLocalTenantMetrics(t.ID)
		_ = s.meta.RevokeTenantAPIKeys(ctx, t.ID)
		writeStatus(w, meta.TenantDeleting)
		return
	}
	if placement != nil {
		if placement.Status != meta.PlacementStatusDeleting {
			if err := s.pool.PurgeSharedTenant(ctx, fsID, placement.DbID); err != nil {
				// Do not roll back the deleting mark; the purge is resumable and the
				// next delete request will continue it.
				logger.Error(ctx, "shared_tenant_purge_failed", zap.String("tenant_id", t.ID), zap.Int64("db_id", placement.DbID), zap.Error(err))
				errJSON(w, backendErrorStatus(r.Context(), err), "failed to purge tenant data from shared db")
				return
			}

			_ = s.meta.AbortActiveUploadReservations(ctx, t.ID)
			markDeleted := t.StorageNamespaceID == ""
			if err := s.meta.FinalizeSharedTenantDeleteMetadata(ctx, t.ID, fsID, placement.DbID, s.sharedDBReopenRatio, markDeleted); err != nil {
				logger.Error(ctx, "shared_tenant_metadata_finalize_failed", zap.String("tenant_id", t.ID), zap.Error(err))
				errJSON(w, backendErrorStatus(r.Context(), err), "failed to finalize shared tenant deletion")
				return
			}
			if markDeleted {
				s.clearLocalTenantMetrics(t.ID)
				_ = s.meta.RevokeTenantAPIKeys(ctx, t.ID)
				writeTenantDeleteStatus(w, meta.TenantDeleted)
				return
			}
		}

		status := meta.TenantDeleting
		if !deleteJobArmed {
			var enqueueErr error
			status, enqueueErr = s.enqueueTenantDeleteJob(ctx, t)
			if enqueueErr != nil {
				errJSON(w, backendErrorStatus(r.Context(), enqueueErr), "failed to enqueue tenant delete cleanup")
				return
			}
		}
		_ = s.meta.RevokeTenantAPIKeys(ctx, t.ID)
		writeTenantDeleteStatus(w, status)
		return
	}
	logger.Error(ctx, "shared_tenant_delete_placement_missing", zap.String("tenant_id", t.ID), zap.Int64("fs_id", fsID))
	errJSON(w, http.StatusInternalServerError, "shared tenant placement is missing")
}

// delayedTenantDeleteJob builds a shared-schema S3 cleanup job with crash
// fallback headroom. The success path makes it immediately due after purge;
// otherwise it becomes eligible only if the request never returns.
func delayedTenantDeleteJob(t *meta.Tenant, ns *meta.StorageNamespace) *meta.TenantDeleteJob {
	return &meta.TenantDeleteJob{
		TenantID:    t.ID,
		NamespaceID: ns.ID,
		Backend:     ns.Backend,
		Bucket:      ns.Bucket,
		Prefix:      ns.Prefix,
		NotBefore:   time.Now().UTC().Add(defaultTenantDeleteJobFallbackDelay),
	}
}

// markTenantDeletingForCleanup starts the durable delete state machine used
// by owner, admin, and shared-schema request paths. Shared-schema tenants
// atomically persist the deleting mark and delayed cleanup job. Cluster-backed
// tenants persist only the deleting mark; their runnable job is created after
// external deprovision succeeds.
func (s *Server) markTenantDeletingForCleanup(ctx context.Context, t *meta.Tenant) (bool, error) {
	if tenant.SupportsClusterDelete(t.Provider) {
		if t.StorageNamespaceID != "" {
			if _, err := s.meta.GetStorageNamespace(ctx, t.StorageNamespaceID); err != nil {
				return false, fmt.Errorf("load tenant storage namespace: %w", err)
			}
		}
		if t.Status == meta.TenantDeleting {
			return true, nil
		}
		return s.meta.UpdateTenantStatusIf(ctx, t.ID, t.Status, meta.TenantDeleting)
	}
	if t.StorageNamespaceID == "" {
		if t.Status == meta.TenantDeleting {
			return true, nil
		}
		return s.meta.UpdateTenantStatusIf(ctx, t.ID, t.Status, meta.TenantDeleting)
	}
	ns, err := s.meta.GetStorageNamespace(ctx, t.StorageNamespaceID)
	if err != nil {
		return false, fmt.Errorf("load tenant storage namespace: %w", err)
	}
	return s.meta.MarkTenantDeletingWithDeleteJob(ctx, t.ID, t.Status, delayedTenantDeleteJob(t, ns))
}

func (s *Server) enqueueTenantDeleteJob(ctx context.Context, t *meta.Tenant) (meta.TenantStatus, error) {
	if t.StorageNamespaceID == "" {
		if err := s.meta.MarkTenantDeleted(ctx, t.ID); err != nil {
			return "", err
		}
		s.clearLocalTenantMetrics(t.ID)
		return meta.TenantDeleted, nil
	}
	ns, err := s.meta.GetStorageNamespace(ctx, t.StorageNamespaceID)
	if err != nil {
		return "", err
	}
	if tenant.SupportsClusterDelete(t.Provider) {
		updated, err := s.meta.MarkTenantDeletingWithDeleteJob(ctx, t.ID, meta.TenantDeleting, &meta.TenantDeleteJob{
			TenantID:    t.ID,
			NamespaceID: ns.ID,
			Backend:     ns.Backend,
			Bucket:      ns.Bucket,
			Prefix:      ns.Prefix,
		})
		if err != nil {
			return "", err
		}
		if !updated {
			current, err := s.meta.GetTenant(ctx, t.ID)
			if err != nil {
				return "", err
			}
			if current.Status != meta.TenantDeleted {
				return "", fmt.Errorf("tenant delete cleanup handoff lost ownership")
			}
			return meta.TenantDeleted, nil
		}
		s.clearLocalTenantMetrics(t.ID)
		return meta.TenantDeleting, nil
	}
	if err := s.meta.UpdateStorageNamespaceState(ctx, ns.ID, meta.StorageNamespaceDeleting); err != nil {
		return "", err
	}
	if err := s.meta.EnqueueTenantDeleteJob(ctx, &meta.TenantDeleteJob{
		TenantID:    t.ID,
		NamespaceID: ns.ID,
		Backend:     ns.Backend,
		Bucket:      ns.Bucket,
		Prefix:      ns.Prefix,
	}); err != nil {
		return "", err
	}
	s.clearLocalTenantMetrics(t.ID)
	return meta.TenantDeleting, nil
}

func writeTenantDeleteStatus(w http.ResponseWriter, status meta.TenantStatus) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	_ = json.NewEncoder(w).Encode(map[string]string{"status": string(status)})
}

func (s *Server) deprovisionTenantCluster(ctx context.Context, t *meta.Tenant, req tenant.CredentialProvisionRequest) error {
	cluster := clusterInfoFromTenant(t)
	switch t.Provider {
	case tenant.ProviderTiDBCloudNative:
		deprovisioner, ok := s.provisioner.(tenant.CredentialDeprovisioner)
		if !ok {
			return fmt.Errorf("provisioner does not support credential deprovision")
		}
		return deprovisioner.DeprovisionWithCredentials(ctx, cluster, req)
	case tenant.ProviderTiDBCloudStarterLegacy:
		deprovisioner, ok := s.legacyStarterProvisioner.(tenant.Deprovisioner)
		if !ok {
			return fmt.Errorf("legacy starter provisioner does not support deprovision")
		}
		return deprovisioner.Deprovision(ctx, cluster)
	case tenant.ProviderLocal:
		provisioner := s.provisionerForTenantProvider(t.Provider)
		if provisioner == nil {
			return fmt.Errorf("provisioner is not configured for provider %s", t.Provider)
		}
		deprovisioner, ok := provisioner.(tenant.Deprovisioner)
		if !ok {
			return fmt.Errorf("provisioner does not support deprovision")
		}
		return deprovisioner.Deprovision(ctx, cluster)
	default:
		return fmt.Errorf("delete is not supported for provider %s", t.Provider)
	}
}

func decodeCredentialDeprovisionRequest(w http.ResponseWriter, r *http.Request) (tenant.CredentialProvisionRequest, error) {
	var raw struct {
		PublicKey  string `json:"public_key"`
		PrivateKey string `json:"private_key"`
	}
	req, err := decodeCredentialRequest(w, r, &raw, func() tenant.CredentialProvisionRequest {
		return tenant.CredentialProvisionRequest{
			PublicKey:  strings.TrimSpace(raw.PublicKey),
			PrivateKey: strings.TrimSpace(raw.PrivateKey),
		}
	})
	if err != nil {
		return tenant.CredentialProvisionRequest{}, err
	}
	return *req, nil
}

func (s *Server) processTenantDeleteJobs(ctx context.Context) {
	now := time.Now().UTC()
	if _, err := s.meta.RecoverStaleTenantDeleteJobs(ctx, now.Add(-defaultTenantDeleteRunningTTL)); err != nil {
		logger.Error(ctx, "tenant_delete_recover_stale_failed", zap.Error(err))
		return
	}
	jobs, err := s.meta.ListDueTenantDeleteJobs(ctx, now, 25)
	if err != nil {
		logger.Error(ctx, "tenant_delete_list_due_failed", zap.Error(err))
		return
	}
	for _, job := range jobs {
		if processErr := s.processTenantDeleteJob(ctx, job); processErr != nil {
			retryAt := time.Now().UTC().Add(defaultTenantDeleteRetryDelay)
			if retryErr := s.meta.RetryTenantDeleteJob(ctx, job.TenantID, retryAt, processErr.Error()); retryErr != nil {
				logger.Error(ctx, "tenant_delete_job_retry_failed",
					zap.String("tenant_id", job.TenantID),
					zap.Error(processErr),
					zap.NamedError("retry_error", retryErr))
				continue
			}
			logger.Warn(ctx, "tenant_delete_job_retry_scheduled",
				zap.String("tenant_id", job.TenantID),
				zap.Int("attempt", job.Attempts+1),
				zap.Time("retry_at", retryAt),
				zap.Error(processErr))
		}
	}
}

func (s *Server) processTenantDeleteJob(ctx context.Context, job meta.TenantDeleteJob) error {
	updated, err := s.meta.MarkTenantDeleteJobRunning(ctx, job.TenantID)
	if err != nil {
		return err
	}
	if !updated {
		return nil
	}
	// Defense in depth: never run the destructive S3 cleanup for a tenant that
	// is no longer deleting; another path may have rolled its delete back.
	t, tenantErr := s.meta.GetTenant(ctx, job.TenantID)
	if tenantErr != nil && !errors.Is(tenantErr, meta.ErrNotFound) {
		return tenantErr
	}
	if tenantErr != nil || t.Status != meta.TenantDeleting {
		reason := "tenant not deleting"
		if errors.Is(tenantErr, meta.ErrNotFound) {
			reason = "tenant missing"
		}
		cancelled, err := s.meta.CancelTenantDeleteJob(ctx, job.TenantID)
		if err != nil {
			return err
		}
		if cancelled {
			logger.Warn(ctx, "tenant_delete_job_cancelled",
				zap.String("tenant_id", job.TenantID),
				zap.String("reason", reason))
			return nil
		}
		// A concurrent delete reached its durable mark before cancellation.
		// Release this claim through the normal retry path; the request must
		// finish its external steps before arming object cleanup.
		return fmt.Errorf("tenant delete resumed while stale cleanup was being cancelled")
	}
	if tenant.IsSharedSchemaProvider(t.Provider) {
		fsID, err := s.meta.ResolveFsID(ctx, t.ID)
		if err != nil {
			return fmt.Errorf("resolve shared tenant fs_id before object cleanup: %w", err)
		}
		placement, err := s.meta.GetTenantPlacement(ctx, fsID)
		if err != nil {
			return fmt.Errorf("load shared tenant placement before object cleanup: %w", err)
		}
		if placement.Status != meta.PlacementStatusDeleting {
			return fmt.Errorf("shared tenant purge is incomplete: placement status %q", placement.Status)
		}
	}
	ns := &meta.StorageNamespace{
		ID:            job.NamespaceID,
		OwnerTenantID: job.TenantID,
		Backend:       job.Backend,
		Bucket:        job.Bucket,
		Prefix:        job.Prefix,
		State:         meta.StorageNamespaceDeleting,
	}
	s3c, err := s.pool.S3ForStorageNamespace(ctx, ns)
	if err != nil {
		return err
	}
	jobCtx, cancel := context.WithTimeout(ctx, defaultTenantDeleteJobTimeout)
	defer cancel()
	// The S3 client is already scoped to job.Prefix, so an empty relative
	// prefix deletes the whole storage namespace without opening tenant DB.
	res, err := s3c.DeletePrefix(jobCtx, "")
	if err != nil {
		return err
	}
	if job.Backend == "s3" {
		express, err := s.appendLogCleanupClient(jobCtx, &job)
		if err != nil {
			return fmt.Errorf("create append-log cleanup client: %w", err)
		}
		if express != nil {
			expressRes, err := express.DeletePrefix(jobCtx, "")
			if err != nil {
				return fmt.Errorf("delete append-log namespace: %w", err)
			}
			res.DeletedObjects += expressRes.DeletedObjects
			res.AbortedMultipartUploads += expressRes.AbortedMultipartUploads
		}
	}
	if err := s.meta.FinalizeTenantDelete(ctx, job.TenantID, job.NamespaceID, res.DeletedObjects, res.AbortedMultipartUploads); err != nil {
		return err
	}
	s.clearLocalTenantMetrics(job.TenantID)
	return nil
}

func (s *Server) appendLogCleanupClient(ctx context.Context, job *meta.TenantDeleteJob) (s3client.S3Client, error) {
	if job == nil || job.Backend != "s3" {
		return nil, nil
	}
	boundNamespace, err := s.meta.GetStorageNamespace(ctx, job.NamespaceID)
	if err != nil {
		return nil, fmt.Errorf("load append-log storage namespace: %w", err)
	}
	if boundNamespace.OwnerTenantID != job.TenantID {
		return nil, fmt.Errorf("append-log storage namespace %q owner %q does not match tenant %q", boundNamespace.ID, boundNamespace.OwnerTenantID, job.TenantID)
	}
	return s.pool.S3ExpressForAppendLogNamespace(ctx, boundNamespace)
}