feat(jobs): atomic team-deletion teardown + orphan-sweep reconciler

Make team/account deletion eventually-consistent across every system the
provisioner touches. Two changes:

team_deletion_executor:
- Step 0 flips deletion_requested -> deletion_pending the instant teardown
  begins, so a mid-pipeline crash leaves the team in deletion_pending
  (recoverable) rather than indistinguishable from a grace-window team or
  half-tombstoned. The tombstone flip now guards on deletion_pending.
- New step 3: delete every instant-deploy-<appID> k8s namespace the team
  owns. Previously a deleted Pro/Team customer's pods ran (and cost) forever
  — only customer DBs + S3 backups were torn down.
- fetchCandidates also re-sweeps deletion_pending rows for retry; every
  step is idempotent so a re-run after a partial failure completes cleanly.

orphan_sweep_reconciler (new, every 15m):
- PASS 1 finishes stuck deletion_pending teams via the executor seam.
- PASS 2 cancels Razorpay subscriptions still live on tombstoned /
  deletion_pending teams — the "stop the money" backstop so a deleted
  customer is never billed.
- PASS 3 deletes instant-deploy-* namespaces with no live owner.
- Each reclaimed orphan emits team.orphan_reclaimed; each failure emits
  team.orphan_sweep_failed for operator alerting.

K8sNamespaceClient + razorpayOrphanCanceler provide the concrete deleter
and canceler; both fail open (nil) when the cluster / Razorpay is
unreachable.

Tests (sqlmock, no DB needed, run under go test ./...):
- TestTeamDeletionExecutor_TxFailure_StaysPending — mid-deletion failure
  leaves the team in deletion_pending, not half-tombstoned.
- TestTeamDeletionExecutor_PendingRetry_Idempotent — re-running over a
  deletion_pending team completes cleanly.
- TestTeamDeletionExecutor_DeletesK8sNamespaces — namespace teardown.
- TestOrphanSweep_Pass1_CompletesStuckPendingTeam / _TeardownFailure_*
- TestOrphanSweep_Pass2_CancelsOrphanedSubscription / _CancelFailure_*
- TestOrphanSweep_Pass3_DeletesOrphanedNamespaceKeepsLive
- TestOrphanSweep_AllPasses_NoOrphans_Idempotent

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mastermanas805

internal/jobs/k8s_namespace_client.go

@@ -0,0 +1,111 @@
+package jobs
+
+// k8s_namespace_client.go — concrete K8sNamespaceDeleter backed by a
+// kubernetes.Clientset. Shared by the team_deletion_executor (which deletes
+// a deleted team's deploy namespaces) and the orphan_sweep_reconciler
+// (which detects + reclaims orphaned namespaces).
+//
+// WHY ITS OWN FILE
+//
+// deploy_status_reconcile.go already builds a kubernetes.Clientset for its
+// read-only GetDeployment surface. The deletion path needs a *write*
+// surface (Namespaces().Delete) plus a List surface for orphan detection.
+// Keeping the deleter in its own file makes the blast radius of the write
+// capability explicit and gives the orphan-sweep reconciler something to
+// import without dragging in the status-reconciler's autopsy machinery.
+//
+// FAIL-OPEN POSTURE
+//
+// NewK8sNamespaceClient returns (nil, err) when no cluster is reachable
+// (CI, docker-compose). Callers pass nil to the executor / reconciler,
+// which then skip the k8s steps with a WARN log — the same posture as
+// deployStatusK8sProvider. A worker that cannot reach k8s still runs every
+// other periodic job.
+//
+// IDEMPOTENCY
+//
+// DeleteNamespace swallows apierrors.IsNotFound — a namespace that is
+// already gone (a previous partially-failed teardown deleted it, or the
+// deploy-expirer beat us to it) is success, not an error. This is the
+// property that makes re-running a failed team deletion safe.
+
+import (
+	"context"
+	"fmt"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+// k8sNamespaceClient is the concrete K8sNamespaceDeleter. Wraps a
+// kubernetes.Clientset.
+type k8sNamespaceClient struct {
+	cs *kubernetes.Clientset
+}
+
+// NewK8sNamespaceClient builds a K8sNamespaceDeleter from in-cluster config,
+// falling back to the default kubeconfig for local dev. Returns (nil, err)
+// when neither is reachable — the caller logs and passes nil to the
+// executor / reconciler.
+func NewK8sNamespaceClient() (K8sNamespaceDeleter, error) {
+	cs, err := newDeployK8sClientset()
+	if err != nil {
+		return nil, err
+	}
+	return &k8sNamespaceClient{cs: cs}, nil
+}
+
+// DeleteNamespace removes the namespace and everything it contains. A
+// NotFound namespace is treated as success — that is the idempotency
+// contract the executor and the reconciler depend on for safe re-runs.
+//
+// The Delete call is asynchronous on the k8s side (the namespace enters
+// Terminating and the control plane garbage-collects its contents); we do
+// not block on completion. The orphan-sweep reconciler's NamespaceExists
+// check will still report true for a Terminating namespace, so a namespace
+// mid-termination is simply re-observed on the next sweep and re-Delete'd
+// (also a no-op) — eventually consistent, never wrong.
+func (c *k8sNamespaceClient) DeleteNamespace(ctx context.Context, namespace string) error {
+	err := c.cs.CoreV1().Namespaces().Delete(ctx, namespace, metav1.DeleteOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return fmt.Errorf("k8sNamespaceClient.DeleteNamespace %q: %w", namespace, err)
+	}
+	return nil
+}
+
+// NamespaceExists reports whether the namespace is still present (including
+// the Terminating phase). NotFound → (false, nil). Any other error is
+// surfaced so the reconciler does not mistake an API outage for "orphan
+// already cleaned".
+func (c *k8sNamespaceClient) NamespaceExists(ctx context.Context, namespace string) (bool, error) {
+	_, err := c.cs.CoreV1().Namespaces().Get(ctx, namespace, metav1.GetOptions{})
+	if err == nil {
+		return true, nil
+	}
+	if apierrors.IsNotFound(err) {
+		return false, nil
+	}
+	return false, fmt.Errorf("k8sNamespaceClient.NamespaceExists %q: %w", namespace, err)
+}
+
+// ListDeployNamespaces returns the names of every instant-deploy-* namespace
+// currently in the cluster. The orphan-sweep reconciler uses this to find
+// namespaces whose owning deployment row / team is gone. Returns an error
+// on any API failure so the reconciler skips the k8s-orphan phase rather
+// than acting on a truncated list.
+func (c *k8sNamespaceClient) ListDeployNamespaces(ctx context.Context) ([]string, error) {
+	list, err := c.cs.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("k8sNamespaceClient.ListDeployNamespaces: %w", err)
+	}
+	var out []string
+	for i := range list.Items {
+		name := list.Items[i].Name
+		if len(name) >= len(deployNamespacePrefixTDE) &&
+			name[:len(deployNamespacePrefixTDE)] == deployNamespacePrefixTDE {
+			out = append(out, name)
+		}
+	}
+	return out, nil
+}

internal/jobs/orphan_sweep_canceler.go

@@ -0,0 +1,114 @@
+package jobs
+
+// orphan_sweep_canceler.go — the production OrphanSubscriptionCanceler.
+//
+// The orphan-sweep reconciler's PASS 2 cancels Razorpay subscriptions that
+// are still live for a team that has been tombstoned. The worker already
+// links the Razorpay Go SDK (billing_reconciler.go uses it for the
+// subscription fetcher), so we add a thin Cancel wrapper here rather than
+// routing a cancel call through the api over HTTP — fewer hops, and the
+// reconciler is the system of record for "this subscription must die".
+//
+// IDEMPOTENCY
+//
+// Razorpay's cancel endpoint returns an error for a subscription that is
+// already in a terminal state (cancelled / completed / expired). The
+// reconciler must treat "already cancelled" as success or it would loop on
+// the same orphan forever. razorpayOrphanCanceler.CancelSubscription
+// inspects the error text and returns nil for the terminal-state cases —
+// the same already-gone-is-success contract DeleteNamespace uses.
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+
+	razorpay "github.com/razorpay/razorpay-go"
+)
+
+// razorpaySubCancelClient is the subset of razorpay.Client the canceler
+// needs. Narrow interface so tests inject a fake without an httptest
+// server. Mirrors razorpaySDKClient in billing_reconciler.go.
+type razorpaySubCancelClient interface {
+	CancelSubscription(subID string, data map[string]interface{}, extraHeaders map[string]string) (map[string]interface{}, error)
+}
+
+// razorpaySubCancelAdapter wraps razorpay.Client to satisfy the interface.
+type razorpaySubCancelAdapter struct{ c *razorpay.Client }
+
+func (a *razorpaySubCancelAdapter) CancelSubscription(subID string, data map[string]interface{}, extraHeaders map[string]string) (map[string]interface{}, error) {
+	return a.c.Subscription.Cancel(subID, data, extraHeaders)
+}
+
+// razorpayOrphanCanceler implements OrphanSubscriptionCanceler.
+type razorpayOrphanCanceler struct {
+	client razorpaySubCancelClient
+}
+
+// NewRazorpayOrphanCanceler constructs the canceler from RAZORPAY_KEY_ID /
+// RAZORPAY_KEY_SECRET. Returns (nil, nil) when Razorpay is unconfigured so
+// the caller can pass nil into the reconciler — PASS 2 is then skipped with
+// a WARN. Same unconfigured contract as NewRazorpaySubFetcher.
+func NewRazorpayOrphanCanceler() (OrphanSubscriptionCanceler, error) {
+	keyID := os.Getenv("RAZORPAY_KEY_ID")
+	keySecret := os.Getenv("RAZORPAY_KEY_SECRET")
+	if keyID == "" || keySecret == "" {
+		return nil, nil // unconfigured — reconciler skips PASS 2
+	}
+	c := razorpay.NewClient(keyID, keySecret)
+	return &razorpayOrphanCanceler{client: &razorpaySubCancelAdapter{c: c}}, nil
+}
+
+// newRazorpayOrphanCancelerFromClient builds a canceler from a pre-built
+// client. Used by tests to inject a fake.
+func newRazorpayOrphanCancelerFromClient(c razorpaySubCancelClient) *razorpayOrphanCanceler {
+	return &razorpayOrphanCanceler{client: c}
+}
+
+// CancelSubscription issues an immediate (cancel_at_cycle_end=0) cancel.
+//
+// A subscription already in a terminal state is treated as success — the
+// reconciler's goal is "this subscription is not charging the card", and an
+// already-cancelled subscription satisfies that. Without this, the orphan
+// would be re-detected and re-attempted on every sweep forever.
+func (rc *razorpayOrphanCanceler) CancelSubscription(_ context.Context, subscriptionID string) error {
+	if strings.TrimSpace(subscriptionID) == "" {
+		// Nothing to cancel — vacuously satisfied.
+		return nil
+	}
+	_, err := rc.client.CancelSubscription(subscriptionID,
+		map[string]interface{}{"cancel_at_cycle_end": 0}, nil)
+	if err == nil {
+		return nil
+	}
+	if isRazorpayTerminalCancelError(err) {
+		// Already cancelled / completed / expired — the money is already
+		// stopped, which is exactly the post-condition we want.
+		return nil
+	}
+	return fmt.Errorf("razorpayOrphanCanceler.CancelSubscription %q: %w", subscriptionID, err)
+}
+
+// isRazorpayTerminalCancelError reports whether a Razorpay cancel error
+// means the subscription is already in a non-charging terminal state.
+// Razorpay returns a 400 with a message like "subscription is not in a
+// valid state to perform this operation" / "...already been cancelled".
+// We match on the substrings rather than parse the JSON body because the
+// SDK surfaces the error as an opaque error value.
+func isRazorpayTerminalCancelError(err error) bool {
+	msg := strings.ToLower(err.Error())
+	for _, frag := range []string{
+		"already been cancelled",
+		"already cancelled",
+		"not in a valid state",
+		"cannot be cancelled",
+		"completed",
+		"expired",
+	} {
+		if strings.Contains(msg, frag) {
+			return true
+		}
+	}
+	return false
+}