feat: install noncurrent lifecycle schema

migrations/multitenant/0031-lifecycle-tenants.sql

@@ -0,0 +1,71 @@
+CREATE TABLE IF NOT EXISTS public.lifecycle_tenants (
+    tenant_id text PRIMARY KEY
+      REFERENCES public.tenants(id)
+      ON DELETE CASCADE,
+
+    next_dispatch_at timestamptz NOT NULL DEFAULT now(),
+
+    claim_id uuid,
+    claim_until timestamptz,
+
+    last_dispatched_at timestamptz,
+    last_completed_at timestamptz,
+    last_success_at timestamptz,
+
+    failure_count integer NOT NULL DEFAULT 0,
+    last_error jsonb,
+
+    -- Advisory snapshots collected by tenant dispatchers.
+    -- Timestamps expose stale or unavailable tenant observations.
+    -- Scheduling continues to use the tenant state.
+    max_bucket_failure_count integer,
+    oldest_in_flight_started_at timestamptz,
+    bucket_status_observed_at timestamptz,
+
+    CONSTRAINT lifecycle_tenants_failure_count_check
+      CHECK (failure_count >= 0)
+);
+
+CREATE INDEX IF NOT EXISTS lifecycle_tenants_due_idx
+ON public.lifecycle_tenants (
+    next_dispatch_at,
+    tenant_id
+);
+
+CREATE INDEX IF NOT EXISTS lifecycle_tenants_expired_claim_idx
+ON public.lifecycle_tenants (
+    claim_until,
+    tenant_id
+)
+WHERE claim_until IS NOT NULL;
+
+-- Existing policies, restored tenants and freshly migrated tenant DBs need dispatch without another policy PUT.
+CREATE OR REPLACE FUNCTION public.enroll_lifecycle_tenant()
+RETURNS trigger
+LANGUAGE plpgsql
+SET search_path = pg_catalog
+AS $$
+BEGIN
+    IF TG_OP = 'UPDATE'
+        AND OLD.feature_object_versioning IS NOT DISTINCT FROM NEW.feature_object_versioning
+        AND OLD.migrations_version IS NOT DISTINCT FROM NEW.migrations_version THEN
+        RETURN NEW;
+    END IF;
+    INSERT INTO public.lifecycle_tenants (tenant_id, next_dispatch_at)
+    VALUES (NEW.id, clock_timestamp())
+    ON CONFLICT (tenant_id) DO UPDATE
+    SET next_dispatch_at = LEAST(lifecycle_tenants.next_dispatch_at, EXCLUDED.next_dispatch_at);
+    RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS enroll_lifecycle_tenant ON public.tenants;
+CREATE TRIGGER enroll_lifecycle_tenant
+AFTER INSERT OR UPDATE OF feature_object_versioning, migrations_version ON public.tenants
+FOR EACH ROW WHEN (NEW.feature_object_versioning)
+EXECUTE FUNCTION public.enroll_lifecycle_tenant();
+
+INSERT INTO public.lifecycle_tenants (tenant_id, next_dispatch_at)
+SELECT id, clock_timestamp() FROM public.tenants WHERE feature_object_versioning
+ON CONFLICT (tenant_id) DO UPDATE
+SET next_dispatch_at = LEAST(lifecycle_tenants.next_dispatch_at, EXCLUDED.next_dispatch_at);

migrations/tenant/0076-noncurrent-lifecycle.sql

@@ -0,0 +1,86 @@
+-- Add durable lifecycle execution state with one row per bucket.
+
+CREATE TABLE IF NOT EXISTS storage.bucket_lifecycle_states (
+    bucket_id text PRIMARY KEY,
+    generation uuid,
+    next_run_at timestamptz,
+    claim_id uuid,
+    claim_until timestamptz,
+    continuation jsonb,
+    last_started_at timestamptz,
+    last_completed_at timestamptz,
+    last_success_at timestamptz,
+    last_result jsonb,
+    failure_count integer NOT NULL DEFAULT 0,
+    last_error jsonb,
+    updated_at timestamptz NOT NULL DEFAULT now(),
+
+    CONSTRAINT bucket_lifecycle_states_bucket_id_fkey
+      FOREIGN KEY (bucket_id)
+      REFERENCES storage.buckets(id)
+      ON DELETE CASCADE,
+
+    CONSTRAINT bucket_lifecycle_failure_count_check
+      CHECK (failure_count >= 0),
+
+    CONSTRAINT bucket_lifecycle_continuation_object_check
+      CHECK (
+        continuation IS NULL
+        OR jsonb_typeof(continuation) = 'object'
+      ),
+
+    CONSTRAINT bucket_lifecycle_recovery_due_check
+      CHECK (
+        next_run_at IS NOT NULL
+        OR NOT jsonb_path_exists(
+          COALESCE(continuation, '{}'::jsonb),
+          '$.batch.inFlight'
+        )
+      ),
+
+    CONSTRAINT bucket_lifecycle_last_result_object_check
+      CHECK (
+        last_result IS NULL
+        OR jsonb_typeof(last_result) = 'object'
+      )
+);
+
+DO $$
+BEGIN
+    IF NOT (SELECT relrowsecurity FROM pg_catalog.pg_class
+            WHERE oid = 'storage.bucket_lifecycle_states'::regclass) THEN
+        ALTER TABLE storage.bucket_lifecycle_states ENABLE ROW LEVEL SECURITY;
+    END IF;
+    IF to_regclass('storage.bucket_lifecycle_states_due_idx') IS NULL THEN
+        CREATE INDEX IF NOT EXISTS bucket_lifecycle_states_due_idx
+        ON storage.bucket_lifecycle_states (next_run_at, bucket_id)
+        WHERE next_run_at IS NOT NULL;
+    END IF;
+END;
+$$;
+
+DO $$
+DECLARE
+    anon_role text = COALESCE(current_setting('storage.anon_role', true), 'anon');
+    authenticated_role text = COALESCE(current_setting('storage.authenticated_role', true), 'authenticated');
+    service_role text = COALESCE(current_setting('storage.service_role', true), 'service_role');
+    super_user text = COALESCE(current_setting('storage.super_user', true), 'postgres');
+    role_name text;
+BEGIN
+    REVOKE ALL ON TABLE storage.bucket_lifecycle_states FROM PUBLIC;
+
+    FOREACH role_name IN ARRAY ARRAY[anon_role, authenticated_role]
+    LOOP
+      IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN
+        EXECUTE format('REVOKE ALL ON TABLE storage.bucket_lifecycle_states FROM %I', role_name);
+      END IF;
+    END LOOP;
+
+    FOREACH role_name IN ARRAY ARRAY['postgres', super_user, service_role]
+    LOOP
+      IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN
+        EXECUTE format('GRANT ALL ON TABLE storage.bucket_lifecycle_states TO %I', role_name);
+      END IF;
+    END LOOP;
+END;
+$$;

migrations/tenant/0077-noncurrent-lifecycle-archived-order-index.sql

@@ -0,0 +1,5 @@
+-- postgres-migrations disable-transaction
+
+CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS objects_archived_order_uq
+ON storage.objects (bucket_id, name COLLATE "C", archived_at)
+WHERE archived_at IS NOT NULL;

migrations/tenant/0078-noncurrent-lifecycle-archived-due-index.sql

@@ -0,0 +1,10 @@
+-- postgres-migrations disable-transaction
+
+CREATE INDEX CONCURRENTLY IF NOT EXISTS objects_archived_due_idx
+ON storage.objects (
+    bucket_id,
+    archived_at,
+    name COLLATE "C"
+)
+INCLUDE (id, version, is_delete_marker)
+WHERE archived_at IS NOT NULL;