-- =====================================================================
-- Posthawk self-hosted schema bundle
-- =====================================================================
-- Single-paste schema for self-hosters. Apply this to a fresh Supabase
-- project (Settings -> SQL Editor -> paste -> Run). Cloud-only features
-- (Stripe billing, AI assistant, multi-tenant SES, HQ admin tools) are
-- excluded -- set NEXT_PUBLIC_EDITION=self-hosted on the web app and
-- EDITION=self-hosted on the worker so the app knows to hide them.
--
-- Generated from: supabase/migrations/*
-- Included: cloud_domains (unified Domains UI), mcp_auth_codes +
-- mcp_oauth_clients (MCP tables ship in case a future toggle re-enables
-- it — the feature itself is hidden in self-host today via the
-- /api/mcp middleware gate).
-- Excluded tables: workspace_ses_tenants, plans, subscriptions,
-- processed_billing_webhooks, ai_chat_sessions, ai_usage, abuse_alerts,
-- sent_customer_warnings, admin_audit_log, platform_admins,
-- reserved_emails, waitlist_emails, newsletter_emails,
-- admin_login_history RPC, add_platform_admin RPCs (HQ-only),
-- subscription_cancellation_* columns (cloud billing).
--
-- Required Supabase Vault secret AFTER running this:
--   encryption_key  -- used by the worker to encrypt inbound_domains
--                      webhook_secret values at rest. Create it under
--                      Project Settings -> Vault -> Secrets, name
--                      "encryption_key", value: any random 32+ char
--                      string.
--
-- Idempotency: every CREATE statement is "IF NOT EXISTS" where Postgres
-- supports it. The file is NOT safe to re-run after manual schema edits,
-- but is safe to re-run on a fresh project.
-- =====================================================================


-- =====================================================================
-- SECTION: extensions
-- =====================================================================
-- Supabase auto-installs pgcrypto, uuid-ossp, and the vault extension on
-- new projects. We rely on gen_random_uuid() (pgcrypto / pg 14+) and the
-- vault.* schema (Supabase Vault) below.


-- =====================================================================
-- SECTION: utility functions (must come before tables that use them)
-- =====================================================================

-- Auto-update updated_at timestamp
CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS trigger AS $$
BEGIN
  NEW.updated_at = now();
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION update_workspace_updated_at()
RETURNS trigger AS $$
BEGIN
  NEW.updated_at = now();
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Auto-create profile on signup
CREATE OR REPLACE FUNCTION handle_new_user()
RETURNS trigger AS $$
BEGIN
  INSERT INTO public.profiles (id, email, full_name, onboarding_completed)
  VALUES (
    NEW.id,
    NEW.email,
    COALESCE(NEW.raw_user_meta_data->>'full_name', ''),
    false
  )
  ON CONFLICT (id) DO NOTHING;
  RETURN NEW;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Vault secret helper
CREATE OR REPLACE FUNCTION create_vault_secret(secret_value text, secret_description text DEFAULT NULL)
RETURNS uuid AS $$
BEGIN
  RETURN vault.create_secret(secret_value, NULL, secret_description);
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Workspace helper functions (needed by RLS policies below).
--
-- These are defined BEFORE `workspace_members` exists in the file, so we
-- use `LANGUAGE plpgsql` instead of `LANGUAGE sql`. plpgsql defers
-- reference resolution until the function is first executed, by which
-- time the table will have been created further down. `LANGUAGE sql`
-- validates the body at CREATE time and would fail with
-- `relation "workspace_members" does not exist`.
CREATE OR REPLACE FUNCTION is_workspace_member(p_workspace_id uuid, p_user_id uuid)
RETURNS boolean AS $$
BEGIN
  RETURN EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_id = p_workspace_id AND user_id = p_user_id
  );
END;
$$ LANGUAGE plpgsql SECURITY DEFINER STABLE;

CREATE OR REPLACE FUNCTION is_workspace_admin_or_owner(p_workspace_id uuid, p_user_id uuid)
RETURNS boolean AS $$
BEGIN
  RETURN EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_id = p_workspace_id
      AND user_id = p_user_id
      AND role IN ('owner', 'admin')
  );
END;
$$ LANGUAGE plpgsql SECURITY DEFINER STABLE;

-- Generic updated_at trigger helpers (used by some drift-snapshot tables)
CREATE OR REPLACE FUNCTION set_updated_at_now()
RETURNS trigger AS $$
BEGIN
  NEW.updated_at = now();
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION handle_automations_updated_at()
RETURNS trigger AS $$
BEGIN NEW.updated_at = now(); RETURN NEW; END;
$$ LANGUAGE plpgsql;

-- Normalize emails to lowercase + trimmed on INSERT/UPDATE (newsletter
-- subscribers + contacts use this on a BEFORE trigger).
CREATE OR REPLACE FUNCTION normalize_email_basic()
RETURNS trigger AS $$
BEGIN
  IF NEW.email IS NOT NULL THEN
    NEW.email := lower(trim(NEW.email));
  END IF;
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;


-- =====================================================================
-- SECTION: core tables
-- =====================================================================

-- Profiles (extends auth.users)
CREATE TABLE profiles (
  id uuid PRIMARY KEY REFERENCES auth.users(id),
  email text NOT NULL,
  full_name text,
  company_name text,
  avatar_url text,
  timezone text DEFAULT 'UTC',
  onboarding_completed boolean DEFAULT false,
  referral_source text,
  previous_tool text,
  use_case text,
  email_notifications_enabled boolean DEFAULT true,
  limit_alerts_enabled boolean DEFAULT true,
  billing_notifications_enabled boolean DEFAULT true,
  setup_checklist_dismissed boolean DEFAULT false,
  dashboard_widgets jsonb, -- ordered array of overview widget ids the user chose + arranged; NULL = defaults
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now()
);

-- Workspaces
-- Includes the email-verification provider/secret columns so the
-- get_workspace_verification_key / set_workspace_verification_key RPCs
-- compile. Self-hosters who don't use a verification provider can leave
-- both columns NULL.
CREATE TABLE workspaces (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  name text NOT NULL,
  slug text NOT NULL UNIQUE,
  avatar_url text,
  created_by uuid NOT NULL REFERENCES auth.users(id),
  -- 'native' = Posthawk's own validator gates newsletter sign-ups. NULL = off.
  email_verification_provider text CHECK (email_verification_provider IS NULL OR email_verification_provider = 'native'),
  email_verification_secret_id uuid,
  -- AI email design style: a preset id ('minimal' | 'modern-saas' | ...) or
  -- 'custom'; email_custom_style holds the user's own style-guide markdown
  -- (used when the selection is 'custom'). Fed to the AI generator.
  email_design_style text NOT NULL DEFAULT 'modern-saas',
  email_custom_style text,
  -- The workspace's verified "test inboxes" (up to 5). Test-send features
  -- (template test, broadcast test, newsletter issue test) deliver ONLY to one
  -- of these, never to a free-text recipient. Each address is verified via an
  -- emailed code before it's appended here.
  test_emails text[] NOT NULL DEFAULT '{}',
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

-- Workspace members
CREATE TABLE workspace_members (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  user_id uuid NOT NULL REFERENCES auth.users(id),
  role text NOT NULL CHECK (role IN ('owner', 'admin', 'member')),
  joined_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE(workspace_id, user_id)
);

-- FK from workspace_members.user_id -> profiles.id (for PostgREST embedded queries)
ALTER TABLE workspace_members
  ADD CONSTRAINT workspace_members_user_id_profiles_fkey
  FOREIGN KEY (user_id) REFERENCES profiles(id);

-- Workspace invites
CREATE TABLE workspace_invites (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  email text NOT NULL,
  role text NOT NULL CHECK (role IN ('admin', 'member')),
  token text NOT NULL UNIQUE,
  invited_by uuid NOT NULL REFERENCES auth.users(id),
  status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'accepted', 'expired')),
  expires_at timestamptz NOT NULL DEFAULT (now() + interval '7 days'),
  created_at timestamptz NOT NULL DEFAULT now()
);

-- API Keys
CREATE TABLE api_keys (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  key_prefix text NOT NULL,
  key_hash text NOT NULL,
  name text NOT NULL,
  environment text NOT NULL CHECK (environment IN ('test', 'live')),
  scopes text[] NOT NULL DEFAULT '{full_access}',
  last_used_at timestamptz,
  is_mcp boolean NOT NULL DEFAULT false,
  revoked_at timestamptz,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now(),
  UNIQUE(user_id, name, environment)
);

-- Email logs
CREATE TABLE email_logs (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  -- CASCADE so deleting a workspace removes its full sending history too
  -- (a true "delete everything"), rather than orphaning logs with a null
  -- workspace_id. Matches the cloud DB.
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  api_key_id uuid REFERENCES api_keys(id) ON DELETE SET NULL,
  message_id text NOT NULL,
  from_email text NOT NULL,
  to_emails jsonb NOT NULL,
  cc_emails jsonb,
  bcc_emails jsonb,
  subject text,
  status text NOT NULL CHECK (status IN ('queued', 'processing', 'sent', 'delivered', 'opened', 'bounced', 'complained', 'failed', 'sandbox')),
  aws_status text,
  error_message text,
  tags jsonb DEFAULT '[]'::jsonb,
  metadata jsonb DEFAULT '{}'::jsonb,
  email_body jsonb DEFAULT '{}'::jsonb,
  latest_event text,
  queued_at timestamptz DEFAULT now(),
  sent_at timestamptz,
  delivered_at timestamptz,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now()
);

COMMENT ON COLUMN email_logs.email_body IS 'Stores the email body content in JSON format with html and text fields';

-- Email events
CREATE TABLE email_events (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  email_log_id uuid REFERENCES email_logs(id),
  event_type text NOT NULL,
  event_data jsonb NOT NULL,
  created_at timestamptz DEFAULT now()
);

-- Email templates
-- `required_variables` was added by the production-hardening pass so
-- TemplateService can fail fast when callers omit a required substitution.
CREATE TABLE email_templates (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  name text NOT NULL,
  subject text NOT NULL,
  html_body text,
  text_body text,
  template_type text NOT NULL DEFAULT 'html',
  react_source text,
  variables jsonb DEFAULT '[]'::jsonb,
  required_variables jsonb NOT NULL DEFAULT '[]'::jsonb,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now(),
  UNIQUE(user_id, name)
);

-- Scheduled emails
-- `audit_log` + `idempotency_token` were added by the production-hardening
-- pass for race-safe reschedule/cancel and append-only state-change history.
CREATE TABLE scheduled_emails (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  api_key_id uuid REFERENCES api_keys(id) ON DELETE SET NULL,
  email_log_id uuid REFERENCES email_logs(id),
  bullmq_job_id text NOT NULL,
  from_email text NOT NULL,
  to_emails jsonb NOT NULL,
  cc_emails jsonb,
  bcc_emails jsonb,
  subject text NOT NULL,
  html_body text,
  text_body text,
  scheduled_for timestamptz NOT NULL,
  timezone text DEFAULT 'UTC',
  status text DEFAULT 'scheduled' CHECK (status IN ('scheduled', 'sent', 'cancelled', 'failed')),
  sent_at timestamptz,
  cancelled_at timestamptz,
  error_message text,
  audit_log jsonb NOT NULL DEFAULT '[]'::jsonb,
  idempotency_token uuid NOT NULL DEFAULT gen_random_uuid(),
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now()
);

-- Rate limits
CREATE TABLE rate_limits (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  api_key_id uuid REFERENCES api_keys(id) ON DELETE CASCADE,
  limit_type text NOT NULL CHECK (limit_type IN ('hourly', 'daily', 'monthly')),
  max_emails integer NOT NULL DEFAULT 100,
  current_count integer DEFAULT 0,
  window_start timestamptz NOT NULL,
  window_end timestamptz NOT NULL,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now(),
  UNIQUE(api_key_id, limit_type),
  UNIQUE(user_id, limit_type, window_start)
);

-- Domain health (reputation)
CREATE TABLE domain_health (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  domain text NOT NULL,
  verification_status text CHECK (verification_status IN ('PENDING', 'SUCCESS', 'FAILED', 'NOT_STARTED', 'TEMPORARY_FAILURE')),
  verified_for_sending boolean DEFAULT false,
  dkim_status text CHECK (dkim_status IN ('PENDING', 'SUCCESS', 'FAILED', 'NOT_STARTED', 'TEMPORARY_FAILURE')),
  dkim_signing_enabled boolean DEFAULT false,
  dkim_key_length text,
  dkim_tokens jsonb,
  mail_from_domain text,
  mail_from_status text CHECK (mail_from_status IN ('PENDING', 'SUCCESS', 'FAILED', 'NOT_STARTED', 'TEMPORARY_FAILURE')),
  spf_record_valid boolean,
  dmarc_policy text,
  dmarc_record text,
  last_checked_at timestamptz,
  last_success_at timestamptz,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now(),
  UNIQUE(user_id, domain)
);

-- Reputation metrics
CREATE TABLE reputation_metrics (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  account_status text CHECK (account_status IN ('HEALTHY', 'PROBATION', 'SHUTDOWN')),
  bounce_rate numeric,
  complaint_rate numeric,
  bounce_rate_status text CHECK (bounce_rate_status IN ('HEALTHY', 'PROBATION', 'SHUTDOWN')),
  complaint_rate_status text CHECK (complaint_rate_status IN ('HEALTHY', 'PROBATION', 'SHUTDOWN')),
  emails_sent_24h integer DEFAULT 0,
  emails_sent_7d integer DEFAULT 0,
  bounces_24h integer DEFAULT 0,
  complaints_24h integer DEFAULT 0,
  daily_quota bigint DEFAULT 0,
  max_send_rate numeric DEFAULT 0,
  production_access boolean DEFAULT false,
  sending_enabled boolean DEFAULT false,
  synced_at timestamptz DEFAULT now(),
  created_at timestamptz DEFAULT now()
);

-- Inbound domains
CREATE TABLE inbound_domains (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  domain text NOT NULL,
  verification_status text DEFAULT 'pending' CHECK (verification_status IN ('pending', 'verified', 'failed')),
  verification_token text,
  mx_record_configured boolean DEFAULT false,
  webhook_url text,
  webhook_secret text,
  is_active boolean DEFAULT true,
  emails_received_count integer DEFAULT 0,
  last_email_at timestamptz,
  provider text DEFAULT 'cloudflare' CHECK (provider IN ('cloudflare', 'aws_ses', 'custom')),
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now(),
  UNIQUE(user_id, domain)
);

COMMENT ON COLUMN inbound_domains.provider IS 'The email receiving provider: cloudflare (Email Routing), aws_ses, or custom';

-- Inbound emails
-- inbound_domain_id is nullable on self-hosted (single source per row)
-- so the CHECK constraint just requires a domain reference.
CREATE TABLE inbound_emails (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  inbound_domain_id uuid REFERENCES inbound_domains(id),
  message_id text,
  from_email text NOT NULL,
  from_name text,
  to_emails jsonb NOT NULL,
  cc_emails jsonb,
  subject text,
  text_body text,
  html_body text,
  attachments jsonb DEFAULT '[]'::jsonb,
  headers jsonb,
  raw_s3_key text,
  webhook_status text CHECK (webhook_status IN ('pending', 'delivered', 'failed')),
  webhook_response jsonb,
  webhook_attempts integer DEFAULT 0,
  spf_verdict text,
  dkim_verdict text,
  dmarc_verdict text,
  spam_verdict text,
  virus_verdict text,
  received_at timestamptz DEFAULT now(),
  created_at timestamptz DEFAULT now()
);

-- Template images (for email template editor)
CREATE TABLE template_images (
  id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
  workspace_id UUID REFERENCES workspaces(id) ON DELETE CASCADE,
  file_name TEXT NOT NULL,
  original_name TEXT NOT NULL,
  content_type TEXT NOT NULL,
  size_bytes BIGINT NOT NULL,
  storage_path TEXT NOT NULL,
  public_url TEXT NOT NULL,
  tags TEXT[] DEFAULT '{}',
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- API Request Logs (for debugging V1 API calls)
CREATE TABLE api_request_logs (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  api_key_id uuid REFERENCES api_keys(id) ON DELETE SET NULL,
  method text NOT NULL,
  path text NOT NULL,
  status_code integer NOT NULL,
  request_summary jsonb,
  response_summary jsonb,
  duration_ms integer NOT NULL,
  ip_address text,
  -- Where the request originated. Either 'api' (API-key auth via
  -- ApiKeyGuard), 'dashboard' (user JWT via JwtAuthGuard), or
  -- 'unknown' (defensive fallback — should never appear since the
  -- interceptor short-circuits on missing user_id).
  source text,
  created_at timestamptz DEFAULT now()
);

-- Contacts (audience management)
CREATE TABLE contacts (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  email text NOT NULL,
  name text,
  tags text[] DEFAULT '{}',
  metadata jsonb DEFAULT '{}',
  unsubscribed boolean DEFAULT false,
  -- Provenance written by the contact-import worker on every upsert.
  -- import_id FK is added after contact_imports is created (see below).
  consent_source text,
  consent_timestamp timestamptz,
  import_id uuid,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now()
);

-- Contact imports — CSV upload jobs picked up by the worker.
-- The dashboard's /dashboard/contacts page polls this on every visit
-- via <ImportStatusBanner>, so the table must exist even on installs
-- that haven't actually run an import yet.
CREATE TABLE contact_imports (
  id                          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id                     uuid NOT NULL REFERENCES auth.users(id),
  workspace_id                uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,

  file_name                   text NOT NULL,
  file_size_bytes             bigint NOT NULL,
  storage_path                text NOT NULL,

  column_mapping              jsonb NOT NULL DEFAULT '{}'::jsonb,
  default_tags                text[] NOT NULL DEFAULT '{}',
  skip_header                 boolean NOT NULL DEFAULT true,

  -- Consent audit trail captured at import time.
  consent_certified           boolean NOT NULL DEFAULT false,
  consent_basis               text,
  consent_source_description  text,
  consent_ip                  text,
  consent_user_agent          text,

  status                      text NOT NULL DEFAULT 'pending'
    CHECK (status IN ('pending', 'processing', 'completed', 'failed', 'cancelled')),

  total_rows                  integer,
  processed_rows              integer NOT NULL DEFAULT 0,
  imported_rows               integer NOT NULL DEFAULT 0,
  updated_rows                integer NOT NULL DEFAULT 0,
  invalid_rows                integer NOT NULL DEFAULT 0,
  duplicate_rows              integer NOT NULL DEFAULT 0,
  suppressed_rows             integer NOT NULL DEFAULT 0,

  error_message               text,
  row_errors                  jsonb,

  started_at                  timestamptz,
  completed_at                timestamptz,
  created_at                  timestamptz NOT NULL DEFAULT now(),
  updated_at                  timestamptz NOT NULL DEFAULT now()
);

-- Now that contact_imports exists, wire up contacts.import_id → contact_imports.
ALTER TABLE contacts
  ADD CONSTRAINT contacts_import_id_fkey
  FOREIGN KEY (import_id) REFERENCES contact_imports(id) ON DELETE SET NULL;
CREATE INDEX idx_contacts_import_id ON contacts (import_id);

-- Suppression list (bounces, complaints, manual)
CREATE TABLE suppression_list (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  email text NOT NULL,
  reason text NOT NULL CHECK (reason IN ('hard_bounce', 'soft_bounce', 'complaint', 'manual')),
  source_email_log_id uuid REFERENCES email_logs(id) ON DELETE SET NULL,
  notes text,
  created_at timestamptz DEFAULT now(),
  updated_at timestamptz DEFAULT now()
);

-- Soft-bounce counter (per workspace, per address). Counter resets on a
-- Delivery event; auto-suppress fires after 3 consecutive transient bounces.
CREATE TABLE soft_bounce_tracker (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  email text NOT NULL,
  bounce_count integer NOT NULL DEFAULT 0,
  first_bounce_at timestamptz NOT NULL DEFAULT now(),
  last_bounce_at timestamptz NOT NULL DEFAULT now(),
  last_diagnostic text,
  last_email_log_id uuid REFERENCES email_logs(id) ON DELETE SET NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (workspace_id, email)
);

-- Append-only history of contacts removed because of a suppression event:
-- manually (suppressed an address that was in contacts and chose to remove it)
-- or automatically (hard bounce / complaint auto-removal).
CREATE TABLE contact_removals (
  id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id       uuid REFERENCES auth.users(id) ON DELETE SET NULL,
  workspace_id  uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  email         text NOT NULL,
  name          text,
  tags          text[] NOT NULL DEFAULT '{}',
  -- Mirrors suppression_list.reason; 'manual' = removed via the suppressions
  -- page, the rest are automatic via the bounce/complaint rules.
  reason        text NOT NULL CHECK (reason IN ('hard_bounce', 'soft_bounce', 'complaint', 'manual')),
  notes         text,
  removed_at    timestamptz NOT NULL DEFAULT now(),
  created_at    timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX idx_contact_removals_workspace_removed
  ON contact_removals (workspace_id, removed_at DESC);

-- MCP OAuth: short-lived authorization codes for the dashboard workspace-
-- picker flow before exchange to an API key.
CREATE TABLE mcp_auth_codes (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  code text NOT NULL UNIQUE,
  raw_api_key text NOT NULL,
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  api_key_id uuid NOT NULL REFERENCES api_keys(id) ON DELETE CASCADE,
  client_name text NOT NULL,
  redirect_uri text NOT NULL,
  code_challenge text NOT NULL,
  state text,
  expires_at timestamptz NOT NULL DEFAULT (now() + interval '5 minutes'),
  consumed_at timestamptz,
  created_at timestamptz DEFAULT now()
);

-- MCP OAuth dynamic client registration (RFC 7591)
CREATE TABLE mcp_oauth_clients (
  client_id text NOT NULL PRIMARY KEY,
  client_secret text,
  client_name text,
  redirect_uris jsonb NOT NULL DEFAULT '[]'::jsonb,
  client_id_issued_at bigint NOT NULL,
  client_secret_expires_at bigint,
  created_at timestamptz NOT NULL DEFAULT now()
);

-- MFA backup codes (one-time recovery codes for TOTP)
CREATE TABLE mfa_backup_codes (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  code_hash text NOT NULL,
  used_at timestamptz,
  created_at timestamptz NOT NULL DEFAULT now()
);

-- Per-workspace email-address validation cache (Posthawk's native validator).
-- Decisions are TTL'd via expires_at.
CREATE TABLE email_verification_cache (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  email text NOT NULL,
  provider text NOT NULL,
  decision text NOT NULL,
  category text,
  reason text,
  checked_at timestamptz NOT NULL DEFAULT now(),
  expires_at timestamptz NOT NULL,
  UNIQUE (workspace_id, email)
);


-- =====================================================================
-- SECTION: broadcasts (one-off campaigns to contact lists)
-- =====================================================================

CREATE TABLE broadcasts (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  user_id uuid NOT NULL REFERENCES auth.users(id),
  workspace_id uuid REFERENCES workspaces(id) ON DELETE CASCADE,
  name text NOT NULL,
  from_email text NOT NULL,
  from_name text,
  subject text NOT NULL,
  html_body text,
  text_body text,
  recipient_type text NOT NULL DEFAULT 'all',
  recipient_tag text,
  status text NOT NULL DEFAULT 'draft',
  total_recipients integer NOT NULL DEFAULT 0,
  sent_count integer NOT NULL DEFAULT 0,
  failed_count integer NOT NULL DEFAULT 0,
  -- Future send time while status='scheduled'; the worker's broadcast-scheduler
  -- tick polls every minute and dispatches due rows.
  scheduled_for timestamptz,
  sent_at timestamptz,
  cancelled_at timestamptz,
  audit_log jsonb NOT NULL DEFAULT '[]'::jsonb,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);


-- =====================================================================
-- SECTION: newsletters
-- =====================================================================

CREATE TABLE newsletters (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  user_id uuid NOT NULL REFERENCES auth.users(id),
  slug text NOT NULL UNIQUE,
  name text NOT NULL,
  description text,
  logo_url text,
  accent_color text DEFAULT '#8262ff',
  from_email text,
  from_name text,
  reply_to text,
  custom_domain text,
  paid_enabled boolean NOT NULL DEFAULT false,
  stripe_price_id text,
  status text NOT NULL DEFAULT 'active',
  -- Why the newsletter is paused. 'domain_removed' is set by the domain-disconnect
  -- cascade (from_email is kept so re-adding the domain can rematch & resume).
  paused_reason text CHECK (paused_reason IS NULL OR paused_reason = ANY (ARRAY['manual'::text, 'domain_removed'::text])),
  subscriber_count integer NOT NULL DEFAULT 0,
  pending_count integer NOT NULL DEFAULT 0,
  double_opt_in boolean NOT NULL DEFAULT true,
  headless boolean NOT NULL DEFAULT false,
  confirm_redirect_url text,
  unsubscribe_redirect_url text,
  auto_sync_contacts boolean NOT NULL DEFAULT false,
  email_verification_enabled boolean NOT NULL DEFAULT false,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE newsletter_issues (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  newsletter_id uuid NOT NULL REFERENCES newsletters(id) ON DELETE CASCADE,
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  slug text NOT NULL,
  subject text NOT NULL,
  preheader text,
  html_body text,
  text_body text,
  template_id uuid REFERENCES email_templates(id) ON DELETE SET NULL,
  tier_required text NOT NULL DEFAULT 'free',
  status text NOT NULL DEFAULT 'draft',
  scheduled_for timestamptz,
  sent_at timestamptz,
  recipient_count integer NOT NULL DEFAULT 0,
  -- Eligible-subscriber count snapshotted at send start (denominator for the
  -- live progress bar); cancelled_at + audit_log mirror the broadcast trail.
  total_recipients integer,
  cancelled_at timestamptz,
  audit_log jsonb NOT NULL DEFAULT '[]'::jsonb,
  -- Per-issue mapping of the backing template's variables → filled values
  -- (values may embed subscriber tokens like {{name}}). Empty unless backed by
  -- a template that declares variables.
  template_variables jsonb NOT NULL DEFAULT '{}'::jsonb,
  open_count integer NOT NULL DEFAULT 0,
  click_count integer NOT NULL DEFAULT 0,
  public_archive boolean NOT NULL DEFAULT true,
  react_source text,
  editor_mode text NOT NULL DEFAULT 'html',
  -- Resumable fan-out checkpoint: id of the last subscriber queued. NULL = not
  -- started or completed. Lets a crashed/retried send resume instead of
  -- re-queuing from the start (which would duplicate-email everyone).
  send_cursor text,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (newsletter_id, slug)
);

CREATE TABLE newsletter_subscribers (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  newsletter_id uuid NOT NULL REFERENCES newsletters(id) ON DELETE CASCADE,
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  contact_id uuid REFERENCES contacts(id) ON DELETE SET NULL,
  email text NOT NULL,
  name text,
  status text NOT NULL DEFAULT 'pending',
  tier text NOT NULL DEFAULT 'free',
  confirm_token text,
  unsubscribe_token text NOT NULL,
  stripe_subscription_id text,
  subscribed_ip inet,
  subscribed_user_agent text,
  subscribed_at timestamptz NOT NULL DEFAULT now(),
  confirmed_at timestamptz,
  unsubscribed_at timestamptz,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  source text NOT NULL DEFAULT 'public_form',
  confirm_token_issued_at timestamptz DEFAULT now(),
  UNIQUE (newsletter_id, email)
);

-- Throttle log for the public subscribe endpoint. Used by
-- check_and_log_newsletter_subscribe() to enforce per-IP and per-email caps.
CREATE TABLE newsletter_subscribe_attempts (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  newsletter_id uuid REFERENCES newsletters(id) ON DELETE CASCADE,
  email text NOT NULL,
  ip inet,
  triggered_email boolean NOT NULL DEFAULT false,
  attempted_at timestamptz NOT NULL DEFAULT now(),
  outcome text,
  verification_decision text,
  verification_category text,
  verification_cached boolean
);

-- (newsletter_emails removed from self-host bundle — it's only used by
--  the marketing landing page footer on posthawk.dev, which self-hosters
--  don't run. Keeping it would expose an anon-insert table with no rate
--  limit, which spammers can fill.)


-- =====================================================================
-- SECTION: automations
-- =====================================================================

CREATE TABLE automations (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  user_id uuid NOT NULL REFERENCES auth.users(id),
  name text NOT NULL DEFAULT 'Untitled Automation',
  status text NOT NULL DEFAULT 'draft',
  trigger_type text NOT NULL DEFAULT 'contact_added',
  trigger_config jsonb NOT NULL DEFAULT '{"audience": "all"}'::jsonb,
  paused_reason text CHECK (paused_reason IS NULL OR paused_reason = ANY (ARRAY['plan_downgrade'::text, 'manual'::text, 'domain_removed'::text])),
  activated_at timestamptz,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE automation_steps (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  automation_id uuid NOT NULL REFERENCES automations(id) ON DELETE CASCADE,
  order_index integer NOT NULL,
  type text NOT NULL,
  config jsonb NOT NULL DEFAULT '{}'::jsonb,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE automation_enrollments (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  automation_id uuid NOT NULL REFERENCES automations(id) ON DELETE CASCADE,
  -- FK to contacts (nullable: scheduled / cron-triggered enrollments
  -- have no associated contact). The dashboard's automations page
  -- relies on this FK to follow `contacts(email, name)` via PostgREST
  -- embedded select — without it, PostgREST errors with "Could not
  -- find a relationship between 'automation_enrollments' and 'contacts'".
  contact_id uuid REFERENCES contacts(id) ON DELETE SET NULL,
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  status text NOT NULL DEFAULT 'running',
  current_step_index integer NOT NULL DEFAULT 0,
  next_run_at timestamptz,
  bullmq_job_id text,
  last_error text,
  execution_path jsonb,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  enrolled_at timestamptz NOT NULL DEFAULT now(),
  completed_at timestamptz,
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE automation_events (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  enrollment_id uuid NOT NULL REFERENCES automation_enrollments(id) ON DELETE CASCADE,
  step_id uuid REFERENCES automation_steps(id) ON DELETE SET NULL,
  event_type text NOT NULL,
  payload jsonb NOT NULL DEFAULT '{}'::jsonb,
  occurred_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE automation_test_runs (
  id uuid NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
  automation_id uuid NOT NULL REFERENCES automations(id) ON DELETE CASCADE,
  workspace_id uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  contact_id uuid,
  status text NOT NULL,
  steps jsonb NOT NULL DEFAULT '[]'::jsonb,
  final_contact_state jsonb,
  duration_ms integer NOT NULL DEFAULT 0,
  current_step_index integer NOT NULL DEFAULT 0,
  execution_path jsonb,
  state jsonb,
  next_run_at timestamptz,
  bullmq_job_id text,
  created_at timestamptz NOT NULL DEFAULT now()
);


-- =====================================================================
-- SECTION: webhook endpoints + deliveries (event subscriptions)
-- =====================================================================
-- User-configured HTTP endpoints that receive delivery/bounce/complaint
-- /open/click event notifications. Works in both editions.

CREATE TABLE webhook_endpoints (
  id              uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id    uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  url             text NOT NULL,
  secret          text NOT NULL DEFAULT encode(gen_random_bytes(32), 'hex'),
  events          text[] NOT NULL DEFAULT '{}',
  enabled         boolean NOT NULL DEFAULT true,
  description     text,
  created_at      timestamptz NOT NULL DEFAULT now(),
  updated_at      timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE webhook_deliveries (
  id                    uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  webhook_endpoint_id   uuid NOT NULL REFERENCES webhook_endpoints(id) ON DELETE CASCADE,
  event_type            text NOT NULL,
  payload               jsonb NOT NULL,
  status                text NOT NULL DEFAULT 'pending'
    CHECK (status IN ('pending', 'success', 'failed')),
  attempts              integer NOT NULL DEFAULT 0,
  last_attempt_at       timestamptz,
  response_status       integer,
  response_body         text,
  duration_ms           integer,
  created_at            timestamptz NOT NULL DEFAULT now()
);


-- =====================================================================
-- SECTION: indexes
-- =====================================================================

-- profiles, workspaces, workspace_members
CREATE INDEX idx_workspaces_created_by ON workspaces (created_by);
CREATE INDEX idx_workspaces_slug ON workspaces (slug);
CREATE INDEX idx_workspace_members_user_id ON workspace_members (user_id);
CREATE INDEX idx_workspace_members_workspace_id ON workspace_members (workspace_id);

-- workspace_invites
CREATE INDEX idx_workspace_invites_email ON workspace_invites (email);
CREATE INDEX idx_workspace_invites_token ON workspace_invites (token);
CREATE INDEX idx_workspace_invites_workspace_id ON workspace_invites (workspace_id);

-- api_keys
CREATE INDEX idx_api_keys_prefix ON api_keys (key_prefix) WHERE (revoked_at IS NULL);
CREATE INDEX idx_api_keys_user_id ON api_keys (user_id);
CREATE INDEX idx_api_keys_workspace_id ON api_keys (workspace_id);

-- email_logs
CREATE INDEX idx_email_logs_created_at ON email_logs (created_at DESC);
CREATE INDEX idx_email_logs_message_id ON email_logs (message_id);
CREATE INDEX idx_email_logs_status ON email_logs (status);
CREATE INDEX idx_email_logs_user_created ON email_logs (user_id, created_at DESC);
CREATE INDEX idx_email_logs_user_id ON email_logs (user_id);
CREATE INDEX idx_email_logs_workspace_id ON email_logs (workspace_id);
CREATE INDEX idx_email_logs_workspace_user_created
  ON email_logs (workspace_id, user_id, created_at DESC);
-- Dashboard hot-path indexes (from 20260612002100_dashboard_query_indexes)
CREATE INDEX IF NOT EXISTS idx_email_logs_workspace_status_created
  ON email_logs (workspace_id, status, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_email_logs_metadata_source
  ON email_logs ((metadata->>'source'))
  WHERE metadata->>'source' IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_email_logs_metadata_broadcast
  ON email_logs ((metadata->>'broadcastId'))
  WHERE metadata->>'broadcastId' IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_email_logs_workspace_event_created
  ON email_logs (workspace_id, latest_event, created_at DESC)
  WHERE latest_event IS NOT NULL;

-- email_events
CREATE INDEX idx_email_events_created_at ON email_events (created_at DESC);
CREATE INDEX idx_email_events_log_id ON email_events (email_log_id);
CREATE INDEX idx_email_events_type ON email_events (event_type);

-- email_templates
CREATE INDEX idx_email_templates_user_id ON email_templates (user_id);
CREATE INDEX idx_email_templates_workspace_id ON email_templates (workspace_id);

-- scheduled_emails
CREATE INDEX idx_scheduled_emails_bullmq_job_id ON scheduled_emails (bullmq_job_id);
CREATE INDEX idx_scheduled_emails_scheduled_for ON scheduled_emails (scheduled_for);
CREATE INDEX idx_scheduled_emails_status ON scheduled_emails (status);
CREATE INDEX idx_scheduled_emails_user_id ON scheduled_emails (user_id);
CREATE INDEX idx_scheduled_emails_workspace_id ON scheduled_emails (workspace_id);
CREATE INDEX IF NOT EXISTS idx_scheduled_emails_apikey_status
  ON scheduled_emails (api_key_id, status)
  WHERE status = 'scheduled';
CREATE INDEX IF NOT EXISTS idx_scheduled_emails_workspace_created
  ON scheduled_emails (workspace_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_scheduled_emails_workspace_status_created
  ON scheduled_emails (workspace_id, status, created_at DESC);

-- rate_limits
CREATE INDEX idx_rate_limits_api_key_id ON rate_limits (api_key_id);
CREATE INDEX idx_rate_limits_user_id ON rate_limits (user_id);
CREATE INDEX idx_rate_limits_window ON rate_limits (user_id, limit_type, window_start);
CREATE INDEX idx_rate_limits_workspace_id ON rate_limits (workspace_id);

-- domain_health
CREATE INDEX idx_domain_health_domain ON domain_health (domain);
CREATE INDEX idx_domain_health_last_checked ON domain_health (last_checked_at);
CREATE INDEX idx_domain_health_user_id ON domain_health (user_id);
CREATE INDEX idx_domain_health_workspace_id ON domain_health (workspace_id);

-- reputation_metrics
CREATE INDEX idx_reputation_metrics_synced_at ON reputation_metrics (synced_at);
CREATE INDEX idx_reputation_metrics_user_id ON reputation_metrics (user_id);
CREATE INDEX idx_reputation_metrics_workspace_id ON reputation_metrics (workspace_id);

-- inbound_domains
CREATE INDEX idx_inbound_domains_domain ON inbound_domains (domain);
CREATE INDEX idx_inbound_domains_provider ON inbound_domains (provider);
CREATE INDEX idx_inbound_domains_user_id ON inbound_domains (user_id);
CREATE INDEX idx_inbound_domains_verification ON inbound_domains (verification_status);
CREATE INDEX idx_inbound_domains_workspace_id ON inbound_domains (workspace_id);

-- inbound_emails
CREATE INDEX idx_inbound_emails_domain_id ON inbound_emails (inbound_domain_id);
CREATE INDEX idx_inbound_emails_from ON inbound_emails (from_email);
CREATE INDEX idx_inbound_emails_received ON inbound_emails (received_at);
CREATE INDEX idx_inbound_emails_user_id ON inbound_emails (user_id);
CREATE INDEX idx_inbound_emails_webhook_status ON inbound_emails (webhook_status);
CREATE INDEX idx_inbound_emails_workspace_id ON inbound_emails (workspace_id);
CREATE INDEX IF NOT EXISTS idx_inbound_emails_workspace_received
  ON inbound_emails (workspace_id, received_at DESC);
CREATE INDEX IF NOT EXISTS idx_inbound_emails_workspace_domain_received
  ON inbound_emails (workspace_id, inbound_domain_id, received_at DESC)
  WHERE inbound_domain_id IS NOT NULL;

-- template_images
CREATE INDEX idx_template_images_workspace ON template_images (workspace_id);
CREATE INDEX idx_template_images_tags ON template_images USING GIN (tags);

-- api_request_logs
CREATE INDEX idx_api_request_logs_workspace ON api_request_logs (workspace_id, created_at DESC);
CREATE INDEX idx_api_request_logs_api_key ON api_request_logs (api_key_id);

-- contacts
CREATE UNIQUE INDEX idx_contacts_unique ON contacts (workspace_id, email);
CREATE INDEX idx_contacts_workspace ON contacts (workspace_id);
CREATE INDEX idx_contacts_tags ON contacts USING GIN (tags);
CREATE INDEX IF NOT EXISTS idx_contacts_workspace_created
  ON contacts (workspace_id, created_at DESC);

-- contact_imports
CREATE INDEX idx_contact_imports_workspace_created
  ON contact_imports (workspace_id, created_at DESC);
CREATE INDEX idx_contact_imports_status_pending
  ON contact_imports (status)
  WHERE status IN ('pending', 'processing');

-- suppression_list
CREATE UNIQUE INDEX idx_suppression_unique ON suppression_list (workspace_id, email);
CREATE INDEX idx_suppression_workspace ON suppression_list (workspace_id);
CREATE INDEX idx_suppression_email ON suppression_list (email);

-- soft_bounce_tracker
CREATE INDEX idx_soft_bounce_tracker_workspace_email ON soft_bounce_tracker (workspace_id, email);

-- mcp_auth_codes
CREATE INDEX idx_mcp_auth_codes_code ON mcp_auth_codes (code);

-- mfa_backup_codes
CREATE INDEX idx_mfa_backup_codes_user ON mfa_backup_codes (user_id) WHERE used_at IS NULL;

-- email_verification_cache
CREATE INDEX idx_email_verification_cache_workspace_email
  ON email_verification_cache (workspace_id, email);

-- broadcasts
CREATE INDEX idx_broadcasts_workspace_id ON broadcasts (workspace_id);

-- automations
CREATE INDEX idx_automations_workspace_id ON automations (workspace_id);
CREATE INDEX IF NOT EXISTS idx_automations_paused_reason
  ON automations (workspace_id, paused_reason)
  WHERE paused_reason IS NOT NULL;

CREATE INDEX IF NOT EXISTS idx_newsletters_paused_reason
  ON newsletters (workspace_id, paused_reason)
  WHERE paused_reason IS NOT NULL;
CREATE INDEX idx_automation_enrollments_workspace_id ON automation_enrollments (workspace_id);
CREATE INDEX idx_automation_enrollments_automation_id ON automation_enrollments (automation_id);

-- newsletters
CREATE INDEX idx_newsletters_workspace_id ON newsletters (workspace_id);
CREATE INDEX idx_newsletter_issues_newsletter_id ON newsletter_issues (newsletter_id);
CREATE INDEX idx_newsletter_subscribers_newsletter_id ON newsletter_subscribers (newsletter_id);
CREATE INDEX idx_newsletter_subscribers_email ON newsletter_subscribers (email);
CREATE INDEX IF NOT EXISTS idx_newsletter_subscribers_newsletter_status_subscribed
  ON newsletter_subscribers (newsletter_id, status, subscribed_at DESC);
CREATE INDEX IF NOT EXISTS idx_newsletter_subscribers_confirm_token_pending
  ON newsletter_subscribers (confirm_token, confirm_token_issued_at)
  WHERE status = 'pending' AND confirm_token IS NOT NULL;

-- webhook_endpoints
CREATE INDEX idx_webhook_endpoints_workspace_id ON webhook_endpoints (workspace_id);
CREATE INDEX idx_webhook_endpoints_workspace_enabled
  ON webhook_endpoints (workspace_id) WHERE enabled = true;
CREATE INDEX idx_webhook_deliveries_endpoint_created
  ON webhook_deliveries (webhook_endpoint_id, created_at DESC);


-- =====================================================================
-- SECTION: triggers
-- =====================================================================

-- Auto-create profile on auth.users INSERT
CREATE TRIGGER on_auth_user_created
  AFTER INSERT ON auth.users
  FOR EACH ROW EXECUTE FUNCTION handle_new_user();

-- updated_at maintenance
CREATE TRIGGER update_profiles_updated_at
  BEFORE UPDATE ON profiles FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_email_logs_updated_at
  BEFORE UPDATE ON email_logs FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_email_templates_updated_at
  BEFORE UPDATE ON email_templates FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_scheduled_emails_updated_at
  BEFORE UPDATE ON scheduled_emails FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_rate_limits_updated_at
  BEFORE UPDATE ON rate_limits FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_domain_health_updated_at
  BEFORE UPDATE ON domain_health FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_inbound_domains_updated_at
  BEFORE UPDATE ON inbound_domains FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_suppression_list_updated_at
  BEFORE UPDATE ON suppression_list FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_soft_bounce_tracker_updated_at
  BEFORE UPDATE ON soft_bounce_tracker FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_contacts_updated_at
  BEFORE UPDATE ON contacts FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_contact_imports_updated_at
  BEFORE UPDATE ON contact_imports FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_template_images_updated_at
  BEFORE UPDATE ON template_images FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER update_webhook_endpoints_updated_at
  BEFORE UPDATE ON webhook_endpoints FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();
CREATE TRIGGER workspaces_updated_at
  BEFORE UPDATE ON workspaces FOR EACH ROW EXECUTE FUNCTION update_workspace_updated_at();

-- broadcasts / newsletters / automations updated_at
CREATE TRIGGER set_broadcasts_updated_at
  BEFORE UPDATE ON broadcasts FOR EACH ROW EXECUTE FUNCTION set_updated_at_now();
CREATE TRIGGER set_newsletters_updated_at
  BEFORE UPDATE ON newsletters FOR EACH ROW EXECUTE FUNCTION set_updated_at_now();
CREATE TRIGGER set_newsletter_issues_updated_at
  BEFORE UPDATE ON newsletter_issues FOR EACH ROW EXECUTE FUNCTION set_updated_at_now();
CREATE TRIGGER set_automations_updated_at
  BEFORE UPDATE ON automations FOR EACH ROW EXECUTE FUNCTION handle_automations_updated_at();
CREATE TRIGGER set_automation_enrollments_updated_at
  BEFORE UPDATE ON automation_enrollments FOR EACH ROW EXECUTE FUNCTION set_updated_at_now();

-- Normalize email on insert/update for contact-style tables.
CREATE TRIGGER contacts_normalize_email
  BEFORE INSERT OR UPDATE ON contacts FOR EACH ROW EXECUTE FUNCTION normalize_email_basic();
CREATE TRIGGER newsletter_subscribers_normalize_email
  BEFORE INSERT OR UPDATE ON newsletter_subscribers FOR EACH ROW EXECUTE FUNCTION normalize_email_basic();
CREATE TRIGGER suppression_list_normalize_email
  BEFORE INSERT OR UPDATE ON suppression_list FOR EACH ROW EXECUTE FUNCTION normalize_email_basic();

-- Newsletter subscriber count rollup
CREATE OR REPLACE FUNCTION newsletter_subscriber_counts_refresh()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_newsletter_id uuid;
BEGIN
  v_newsletter_id := COALESCE(NEW.newsletter_id, OLD.newsletter_id);
  IF v_newsletter_id IS NULL THEN
    RETURN NULL;
  END IF;
  UPDATE public.newsletters
    SET
      subscriber_count = (
        SELECT count(*)::int FROM public.newsletter_subscribers
        WHERE newsletter_id = v_newsletter_id AND status = 'active'
      ),
      pending_count = (
        SELECT count(*)::int FROM public.newsletter_subscribers
        WHERE newsletter_id = v_newsletter_id AND status = 'pending'
      )
    WHERE id = v_newsletter_id;
  RETURN NULL;
END;
$$;

CREATE TRIGGER newsletter_subscribers_count_refresh
  AFTER INSERT OR UPDATE OR DELETE ON newsletter_subscribers
  FOR EACH ROW EXECUTE FUNCTION newsletter_subscriber_counts_refresh();


-- =====================================================================
-- SECTION: business logic functions / RPCs
-- =====================================================================

-- Rate limit check & increment (atomic)
CREATE OR REPLACE FUNCTION check_and_increment_rate_limit(
  p_api_key_id uuid,
  p_user_id uuid,
  p_limit_type varchar
)
RETURNS boolean AS $$
DECLARE
  v_current_count INTEGER;
  v_max_emails INTEGER;
  v_window_end TIMESTAMPTZ;
  v_window_duration INTERVAL;
BEGIN
  v_window_duration := CASE p_limit_type
    WHEN 'hourly' THEN INTERVAL '1 hour'
    WHEN 'daily' THEN INTERVAL '1 day'
    WHEN 'monthly' THEN INTERVAL '1 month'
  END;

  SELECT current_count, max_emails, window_end INTO v_current_count, v_max_emails, v_window_end
  FROM rate_limits
  WHERE api_key_id = p_api_key_id AND limit_type = p_limit_type
  FOR UPDATE;

  IF NOT FOUND THEN
    INSERT INTO rate_limits (api_key_id, user_id, limit_type, current_count, window_start, window_end)
    VALUES (p_api_key_id, p_user_id, p_limit_type, 1, NOW(), NOW() + v_window_duration);
    RETURN TRUE;
  END IF;

  IF NOW() > v_window_end THEN
    UPDATE rate_limits
    SET current_count = 1,
        window_start = NOW(),
        window_end = NOW() + v_window_duration,
        updated_at = NOW()
    WHERE api_key_id = p_api_key_id AND limit_type = p_limit_type;
    RETURN TRUE;
  END IF;

  IF v_current_count >= v_max_emails THEN
    RETURN FALSE;
  END IF;

  UPDATE rate_limits
  SET current_count = current_count + 1,
      updated_at = NOW()
  WHERE api_key_id = p_api_key_id AND limit_type = p_limit_type;

  RETURN TRUE;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Public invite details (no auth required, used on /invite/[token] page)
CREATE OR REPLACE FUNCTION get_invite_details(invite_token uuid)
RETURNS jsonb AS $$
DECLARE
  v_invite RECORD;
BEGIN
  SELECT wi.*, w.name as workspace_name, p.full_name as inviter_name, p.email as inviter_email
  INTO v_invite
  FROM workspace_invites wi
  JOIN workspaces w ON w.id = wi.workspace_id
  LEFT JOIN profiles p ON p.id = wi.invited_by
  WHERE wi.token = invite_token::text;

  IF NOT FOUND THEN
    RETURN jsonb_build_object('error', 'not_found');
  END IF;

  IF v_invite.status != 'pending' OR v_invite.expires_at < now() THEN
    RETURN jsonb_build_object('error', 'expired');
  END IF;

  RETURN jsonb_build_object(
    'workspace_name', v_invite.workspace_name,
    'inviter_name', COALESCE(v_invite.inviter_name, v_invite.inviter_email, 'A team member'),
    'role', v_invite.role,
    'email', v_invite.email,
    'expires_at', v_invite.expires_at
  );
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Accept workspace invite. Uses the `status` column (no `accepted_at`
-- column required) so the schema stays simple.
CREATE OR REPLACE FUNCTION accept_workspace_invite(invite_token uuid)
RETURNS jsonb AS $$
DECLARE
  v_invite RECORD;
  v_existing RECORD;
  v_workspace_name TEXT;
BEGIN
  SELECT * INTO v_invite
  FROM workspace_invites
  WHERE token = invite_token::text
    AND status = 'pending'
    AND expires_at > now();

  IF NOT FOUND THEN
    RETURN jsonb_build_object('error', 'Invalid or expired invite');
  END IF;

  IF v_invite.email != (auth.jwt() ->> 'email') THEN
    RETURN jsonb_build_object('error', 'This invite was sent to a different email address');
  END IF;

  SELECT id INTO v_existing
  FROM workspace_members
  WHERE workspace_id = v_invite.workspace_id
    AND user_id = auth.uid();

  IF FOUND THEN
    UPDATE workspace_invites SET status = 'accepted' WHERE id = v_invite.id;
    SELECT name INTO v_workspace_name FROM workspaces WHERE id = v_invite.workspace_id;
    RETURN jsonb_build_object('success', true, 'already_member', true, 'workspace_id', v_invite.workspace_id, 'workspace_name', v_workspace_name);
  END IF;

  INSERT INTO workspace_members (workspace_id, user_id, role)
  VALUES (v_invite.workspace_id, auth.uid(), v_invite.role);

  UPDATE workspace_invites SET status = 'accepted' WHERE id = v_invite.id;

  INSERT INTO profiles (id, email, onboarding_completed)
  VALUES (auth.uid(), (auth.jwt() ->> 'email'), true)
  ON CONFLICT (id) DO UPDATE SET onboarding_completed = true;

  SELECT name INTO v_workspace_name FROM workspaces WHERE id = v_invite.workspace_id;
  RETURN jsonb_build_object('success', true, 'already_member', false, 'workspace_id', v_invite.workspace_id, 'workspace_name', v_workspace_name);
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- Total storage bytes used by a workspace's template images
CREATE OR REPLACE FUNCTION get_workspace_storage_bytes(ws_id UUID)
RETURNS BIGINT AS $$
  SELECT COALESCE(SUM(size_bytes), 0) FROM template_images WHERE workspace_id = ws_id;
$$ LANGUAGE sql STABLE SECURITY DEFINER;

-- Soft-bounce auto-suppression helpers. The worker's SuppressionService
-- bumps the counter on every Transient/Undetermined bounce and clears it
-- on Delivery.
CREATE OR REPLACE FUNCTION increment_soft_bounce(
  p_workspace_id uuid,
  p_email text,
  p_diagnostic text DEFAULT NULL,
  p_email_log_id uuid DEFAULT NULL
)
RETURNS integer
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_count integer;
BEGIN
  INSERT INTO public.soft_bounce_tracker (
    workspace_id, email, bounce_count, first_bounce_at, last_bounce_at, last_diagnostic, last_email_log_id
  )
  VALUES (
    p_workspace_id, lower(p_email), 1, now(), now(), p_diagnostic, p_email_log_id
  )
  ON CONFLICT (workspace_id, email) DO UPDATE
    SET bounce_count = soft_bounce_tracker.bounce_count + 1,
        last_bounce_at = now(),
        last_diagnostic = COALESCE(EXCLUDED.last_diagnostic, soft_bounce_tracker.last_diagnostic),
        last_email_log_id = COALESCE(EXCLUDED.last_email_log_id, soft_bounce_tracker.last_email_log_id),
        updated_at = now()
  RETURNING bounce_count INTO v_count;

  RETURN v_count;
END;
$$;

CREATE OR REPLACE FUNCTION clear_soft_bounce(p_workspace_id uuid, p_email text)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  DELETE FROM public.soft_bounce_tracker
  WHERE workspace_id = p_workspace_id
    AND email = lower(p_email);
END;
$$;

-- One-pass engagement + deliverability funnel for a single broadcast, powering
-- the broadcast results modal. SECURITY DEFINER (bypasses RLS) with an explicit
-- membership gate so all workspace members see the same numbers (mirrors
-- get_email_metrics). Fail-closed on a null caller.
--   delivered is cumulative (delivered OR opened — an open implies delivery)
--   opened/clicked come from SES Open/Click events; clicks counted distinct
--   from email_events (status only advances to 'opened', never 'clicked')
CREATE OR REPLACE FUNCTION get_broadcast_stats(p_broadcast_id uuid)
RETURNS TABLE (
  total bigint,
  pending bigint,
  sent bigint,
  delivered bigint,
  opened bigint,
  clicked bigint,
  bounced bigint,
  complained bigint,
  failed bigint,
  delayed bigint,
  first_event_at timestamptz,
  last_event_at timestamptz
)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_workspace_id uuid;
  v_owner uuid;
BEGIN
  IF auth.uid() IS NULL THEN
    RAISE EXCEPTION 'Not authorized';
  END IF;

  SELECT b.workspace_id, b.user_id INTO v_workspace_id, v_owner
  FROM broadcasts b
  WHERE b.id = p_broadcast_id;

  IF NOT FOUND THEN
    RETURN; -- unknown broadcast → no rows
  END IF;

  IF NOT (
    v_owner = auth.uid()
    OR (v_workspace_id IS NOT NULL AND EXISTS (
      SELECT 1 FROM workspace_members wm
      WHERE wm.workspace_id = v_workspace_id AND wm.user_id = auth.uid()
    ))
    OR (v_workspace_id IS NOT NULL AND EXISTS (
      SELECT 1 FROM workspaces w
      WHERE w.id = v_workspace_id AND w.created_by = auth.uid()
    ))
  ) THEN
    RAISE EXCEPTION 'Not authorized to view this broadcast';
  END IF;

  RETURN QUERY
  WITH logs AS (
    SELECT el.id, el.status, el.latest_event, el.created_at, el.updated_at
    FROM email_logs el
    WHERE el.metadata->>'broadcastId' = p_broadcast_id::text
  ),
  clicks AS (
    SELECT count(DISTINCT ee.email_log_id) AS c
    FROM email_events ee
    JOIN logs l ON l.id = ee.email_log_id
    WHERE ee.event_type = 'click'
  )
  SELECT
    count(*)::bigint,
    count(*) FILTER (WHERE status IN ('queued', 'processing'))::bigint,
    count(*) FILTER (WHERE status = 'sent')::bigint,
    count(*) FILTER (WHERE status IN ('delivered', 'opened'))::bigint,
    count(*) FILTER (WHERE status = 'opened')::bigint,
    (SELECT c FROM clicks)::bigint,
    count(*) FILTER (WHERE status = 'bounced')::bigint,
    count(*) FILTER (WHERE status = 'complained')::bigint,
    count(*) FILTER (WHERE status = 'failed')::bigint,
    -- "delayed": accepted by SES, then a DeliveryDelay event (SES retrying).
    -- Subset of status='sent'; UI shows it apart from plain "in transit".
    count(*) FILTER (WHERE status = 'sent' AND latest_event = 'delivery_delay')::bigint,
    min(created_at),
    max(updated_at)
  FROM logs;
END;
$$;

GRANT EXECUTE ON FUNCTION get_broadcast_stats(uuid) TO authenticated;

-- Engagement + deliverability funnel for a single newsletter issue, powering
-- the per-issue reports modal. SECURITY DEFINER with a workspace-membership
-- gate so all members see the same numbers (mirrors get_broadcast_stats).
-- Newsletter fan-out jobs tag email_logs with metadata.issue_id.
CREATE OR REPLACE FUNCTION get_newsletter_issue_stats(p_issue_id uuid)
RETURNS TABLE (
  total bigint,
  pending bigint,
  sent bigint,
  delivered bigint,
  opened bigint,
  clicked bigint,
  bounced bigint,
  complained bigint,
  failed bigint,
  delayed bigint,
  first_event_at timestamptz,
  last_event_at timestamptz
)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_workspace_id uuid;
  v_owner uuid;
BEGIN
  IF auth.uid() IS NULL THEN
    RAISE EXCEPTION 'Not authorized';
  END IF;

  SELECT i.workspace_id, n.user_id INTO v_workspace_id, v_owner
  FROM newsletter_issues i
  JOIN newsletters n ON n.id = i.newsletter_id
  WHERE i.id = p_issue_id;

  IF NOT FOUND THEN
    RETURN; -- unknown issue → no rows
  END IF;

  IF NOT (
    v_owner = auth.uid()
    OR (v_workspace_id IS NOT NULL AND EXISTS (
      SELECT 1 FROM workspace_members wm
      WHERE wm.workspace_id = v_workspace_id AND wm.user_id = auth.uid()
    ))
    OR (v_workspace_id IS NOT NULL AND EXISTS (
      SELECT 1 FROM workspaces w
      WHERE w.id = v_workspace_id AND w.created_by = auth.uid()
    ))
  ) THEN
    RAISE EXCEPTION 'Not authorized to view this issue';
  END IF;

  RETURN QUERY
  WITH logs AS (
    SELECT el.id, el.status, el.latest_event, el.created_at, el.updated_at
    FROM email_logs el
    WHERE el.metadata->>'issue_id' = p_issue_id::text
      -- Real fan-out only; test sends (source='newsletter.issue.test') also
      -- carry issue_id and would otherwise inflate the report.
      AND el.metadata->>'source' = 'newsletter.issue'
  ),
  clicks AS (
    SELECT count(DISTINCT ee.email_log_id) AS c
    FROM email_events ee
    JOIN logs l ON l.id = ee.email_log_id
    WHERE ee.event_type = 'click'
  )
  SELECT
    count(*)::bigint,
    count(*) FILTER (WHERE status IN ('queued', 'processing'))::bigint,
    count(*) FILTER (WHERE status = 'sent')::bigint,
    count(*) FILTER (WHERE status IN ('delivered', 'opened'))::bigint,
    count(*) FILTER (WHERE status = 'opened')::bigint,
    (SELECT c FROM clicks)::bigint,
    count(*) FILTER (WHERE status = 'bounced')::bigint,
    count(*) FILTER (WHERE status = 'complained')::bigint,
    count(*) FILTER (WHERE status = 'failed')::bigint,
    -- "delayed": accepted by SES, then a DeliveryDelay event (SES retrying).
    count(*) FILTER (WHERE status = 'sent' AND latest_event = 'delivery_delay')::bigint,
    min(created_at),
    max(updated_at)
  FROM logs;
END;
$$;

GRANT EXECUTE ON FUNCTION get_newsletter_issue_stats(uuid) TO authenticated;

-- Service-role variant for the API-key worker path (SDK per-issue stats). The
-- worker runs as service_role (auth.uid() is null), so it can't use the gated
-- RPC above — it authorizes via workspace ownership before calling this one.
CREATE OR REPLACE FUNCTION get_newsletter_issue_stats_admin(p_issue_id uuid)
RETURNS TABLE (
  total bigint,
  pending bigint,
  sent bigint,
  delivered bigint,
  opened bigint,
  clicked bigint,
  bounced bigint,
  complained bigint,
  failed bigint,
  delayed bigint,
  first_event_at timestamptz,
  last_event_at timestamptz
)
LANGUAGE sql
SECURITY DEFINER
AS $$
  WITH logs AS (
    SELECT el.id, el.status, el.latest_event, el.created_at, el.updated_at
    FROM email_logs el
    WHERE el.metadata->>'issue_id' = p_issue_id::text
      AND el.metadata->>'source' = 'newsletter.issue'
  ),
  clicks AS (
    SELECT count(DISTINCT ee.email_log_id) AS c
    FROM email_events ee
    JOIN logs l ON l.id = ee.email_log_id
    WHERE ee.event_type = 'click'
  )
  SELECT
    count(*)::bigint,
    count(*) FILTER (WHERE status IN ('queued', 'processing'))::bigint,
    count(*) FILTER (WHERE status = 'sent')::bigint,
    count(*) FILTER (WHERE status IN ('delivered', 'opened'))::bigint,
    count(*) FILTER (WHERE status = 'opened')::bigint,
    (SELECT c FROM clicks)::bigint,
    count(*) FILTER (WHERE status = 'bounced')::bigint,
    count(*) FILTER (WHERE status = 'complained')::bigint,
    count(*) FILTER (WHERE status = 'failed')::bigint,
    count(*) FILTER (WHERE status = 'sent' AND latest_event = 'delivery_delay')::bigint,
    min(created_at),
    max(updated_at)
  FROM logs;
$$;

REVOKE ALL ON FUNCTION get_newsletter_issue_stats_admin(uuid) FROM public;
GRANT EXECUTE ON FUNCTION get_newsletter_issue_stats_admin(uuid) TO service_role;

-- Set-based bulk contact delete. Records every matched contact in
-- contact_removals (reason 'manual') then deletes them — all in a single
-- statement, so it stays fast even for very large result sets. SECURITY
-- DEFINER with an explicit workspace-membership gate (mirrors get_broadcast_stats
-- / get_email_metrics). Requires at least one filter so it can never wipe a
-- whole workspace. p_dry_run returns the match count without deleting.
CREATE OR REPLACE FUNCTION bulk_delete_contacts(
  p_workspace_id uuid,
  p_email_mode   text DEFAULT NULL,
  p_email_value  text DEFAULT NULL,
  p_domain       text DEFAULT NULL,
  p_tags         text[] DEFAULT NULL,
  p_tag_mode     text DEFAULT 'any',
  p_status       text DEFAULT 'all',
  p_notes        text DEFAULT NULL,
  p_dry_run      boolean DEFAULT false
)
RETURNS integer
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_uid            uuid := auth.uid();
  v_email_escaped  text;
  v_email_pattern  text;
  v_domain_pattern text;
  v_has_tags       boolean := p_tags IS NOT NULL AND array_length(p_tags, 1) IS NOT NULL;
  v_count          integer;
BEGIN
  IF v_uid IS NULL THEN
    RAISE EXCEPTION 'Not authorized' USING ERRCODE = '42501';
  END IF;

  -- Membership gate (DEFINER bypasses RLS).
  IF NOT (
    is_workspace_member(p_workspace_id, v_uid)
    OR EXISTS (SELECT 1 FROM workspaces w WHERE w.id = p_workspace_id AND w.created_by = v_uid)
  ) THEN
    RAISE EXCEPTION 'Not a member of workspace %', p_workspace_id USING ERRCODE = '42501';
  END IF;

  -- Build the email-match pattern from the chosen mode.
  IF p_email_value IS NOT NULL AND length(trim(p_email_value)) > 0 THEN
    v_email_escaped := replace(replace(replace(lower(trim(p_email_value)), '\', '\\'), '%', '\%'), '_', '\_');
    v_email_pattern := CASE p_email_mode
      WHEN 'prefix' THEN v_email_escaped || '%'
      WHEN 'suffix' THEN '%' || v_email_escaped
      WHEN 'exact'  THEN v_email_escaped
      ELSE '%' || v_email_escaped || '%'
    END;
  END IF;

  -- Domain match: email ends with @domain.
  IF p_domain IS NOT NULL AND length(trim(p_domain)) > 0 THEN
    v_domain_pattern := '%@' || replace(replace(replace(lower(regexp_replace(trim(p_domain), '^@', '')), '\', '\\'), '%', '\%'), '_', '\_');
  END IF;

  -- Refuse to run with no constraints — never delete an entire workspace.
  IF v_email_pattern IS NULL
     AND v_domain_pattern IS NULL
     AND NOT v_has_tags
     AND (p_status IS NULL OR p_status = 'all') THEN
    RAISE EXCEPTION 'At least one filter is required';
  END IF;

  IF p_dry_run THEN
    SELECT count(*) INTO v_count
    FROM contacts c
    WHERE c.workspace_id = p_workspace_id
      AND (v_email_pattern IS NULL OR c.email ILIKE v_email_pattern)
      AND (v_domain_pattern IS NULL OR c.email ILIKE v_domain_pattern)
      AND (NOT v_has_tags OR (CASE WHEN p_tag_mode = 'all' THEN c.tags @> p_tags ELSE c.tags && p_tags END))
      AND (p_status IS NULL OR p_status = 'all'
           OR (p_status = 'subscribed'   AND c.unsubscribed = false)
           OR (p_status = 'unsubscribed' AND c.unsubscribed = true));
    RETURN COALESCE(v_count, 0);
  END IF;

  WITH matched AS (
    SELECT c.id, c.email, c.name, c.tags
    FROM contacts c
    WHERE c.workspace_id = p_workspace_id
      AND (v_email_pattern IS NULL OR c.email ILIKE v_email_pattern)
      AND (v_domain_pattern IS NULL OR c.email ILIKE v_domain_pattern)
      AND (NOT v_has_tags OR (CASE WHEN p_tag_mode = 'all' THEN c.tags @> p_tags ELSE c.tags && p_tags END))
      AND (p_status IS NULL OR p_status = 'all'
           OR (p_status = 'subscribed'   AND c.unsubscribed = false)
           OR (p_status = 'unsubscribed' AND c.unsubscribed = true))
  ),
  ins AS (
    INSERT INTO contact_removals (user_id, workspace_id, email, name, tags, reason, notes)
    SELECT v_uid, p_workspace_id, m.email, m.name, COALESCE(m.tags, '{}'), 'manual', p_notes
    FROM matched m
  ),
  del AS (
    DELETE FROM contacts WHERE id IN (SELECT id FROM matched) RETURNING 1
  )
  SELECT count(*) INTO v_count FROM del;

  RETURN COALESCE(v_count, 0);
END;
$$;

GRANT EXECUTE ON FUNCTION bulk_delete_contacts(uuid, text, text, text, text[], text, text, text, boolean) TO authenticated;

-- Cascade-delete helper for retention worker. Handles email_events FK +
-- scheduled_emails.email_log_id nullification + email_logs delete in
-- one SQL call instead of 3 HTTP round-trips.
CREATE OR REPLACE FUNCTION cleanup_old_email_logs(
  p_workspace_id uuid,
  p_cutoff timestamptz
)
RETURNS bigint
LANGUAGE plpgsql SECURITY DEFINER
AS $$
DECLARE
  deleted_count bigint;
BEGIN
  DELETE FROM email_events
  WHERE email_log_id IN (
    SELECT id FROM email_logs
    WHERE workspace_id = p_workspace_id
      AND created_at < p_cutoff
  );

  UPDATE scheduled_emails
  SET email_log_id = NULL
  WHERE email_log_id IN (
    SELECT id FROM email_logs
    WHERE workspace_id = p_workspace_id
      AND created_at < p_cutoff
  );

  DELETE FROM email_logs
  WHERE workspace_id = p_workspace_id
    AND created_at < p_cutoff;

  GET DIAGNOSTICS deleted_count = ROW_COUNT;
  RETURN deleted_count;
END;
$$;

-- Newsletter public-subscribe rate-limiter (single call: throttle check +
-- attempt log inside one advisory-locked transaction). Both the legacy
-- 4-arg and the newer 5-arg with attempt_id are defined so older worker
-- builds still resolve.
CREATE OR REPLACE FUNCTION check_and_log_newsletter_subscribe(
  p_newsletter_id uuid,
  p_email text,
  p_ip inet,
  p_will_trigger_email boolean
)
RETURNS TABLE(ok boolean, status_code integer, reason text)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_email text;
  v_ip_count int;
  v_email_10min int;
  v_email_24h int;
  v_lock_key bigint;
BEGIN
  v_email := lower(trim(p_email));
  v_lock_key := hashtextextended(p_newsletter_id::text || '|' || v_email, 0);
  PERFORM pg_advisory_xact_lock(v_lock_key);

  IF p_ip IS NOT NULL THEN
    SELECT count(*) INTO v_ip_count
      FROM newsletter_subscribe_attempts
     WHERE newsletter_id = p_newsletter_id
       AND ip = p_ip
       AND attempted_at > now() - interval '1 hour';
    IF v_ip_count >= 10 THEN
      RETURN QUERY SELECT false, 429,
        'Too many sign-ups from this network. Wait a bit and try again.'::text;
      RETURN;
    END IF;
  END IF;

  SELECT count(*) INTO v_email_10min
    FROM newsletter_subscribe_attempts
   WHERE newsletter_id = p_newsletter_id
     AND email = v_email
     AND triggered_email = true
     AND attempted_at > now() - interval '10 minutes';
  IF v_email_10min >= 1 THEN
    RETURN QUERY SELECT false, 429,
      'A confirmation was just sent to this address. Check your inbox or wait a few minutes.'::text;
    RETURN;
  END IF;

  SELECT count(*) INTO v_email_24h
    FROM newsletter_subscribe_attempts
   WHERE newsletter_id = p_newsletter_id
     AND email = v_email
     AND triggered_email = true
     AND attempted_at > now() - interval '24 hours';
  IF v_email_24h >= 3 THEN
    RETURN QUERY SELECT false, 429,
      'This address has hit today''s confirmation limit for this newsletter. Try again tomorrow.'::text;
    RETURN;
  END IF;

  INSERT INTO newsletter_subscribe_attempts (newsletter_id, email, ip, triggered_email)
  VALUES (p_newsletter_id, v_email, p_ip, p_will_trigger_email);

  RETURN QUERY SELECT true, 200, NULL::text;
END;
$$;

CREATE OR REPLACE FUNCTION check_and_log_newsletter_subscribe(
  p_newsletter_id uuid,
  p_email text,
  p_ip inet,
  p_will_trigger_email boolean,
  p_outcome text
)
RETURNS TABLE(ok boolean, status_code integer, reason text, attempt_id uuid)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_email text;
  v_ip_count int;
  v_email_10min int;
  v_email_24h int;
  v_lock_key bigint;
  v_inserted_id uuid;
BEGIN
  v_email := lower(trim(p_email));
  v_lock_key := hashtextextended(p_newsletter_id::text || '|' || v_email, 0);
  PERFORM pg_advisory_xact_lock(v_lock_key);

  IF p_ip IS NOT NULL THEN
    SELECT count(*) INTO v_ip_count
      FROM newsletter_subscribe_attempts
     WHERE newsletter_id = p_newsletter_id
       AND ip = p_ip
       AND attempted_at > now() - interval '1 hour';
    IF v_ip_count >= 10 THEN
      INSERT INTO newsletter_subscribe_attempts (newsletter_id, email, ip, triggered_email, outcome)
      VALUES (p_newsletter_id, v_email, p_ip, false, 'rejected_rate_limit');
      RETURN QUERY SELECT false, 429,
        'Too many sign-ups from this network. Wait a bit and try again.'::text,
        NULL::uuid;
      RETURN;
    END IF;
  END IF;

  SELECT count(*) INTO v_email_10min
    FROM newsletter_subscribe_attempts
   WHERE newsletter_id = p_newsletter_id
     AND email = v_email
     AND triggered_email = true
     AND attempted_at > now() - interval '10 minutes';
  IF v_email_10min >= 1 THEN
    INSERT INTO newsletter_subscribe_attempts (newsletter_id, email, ip, triggered_email, outcome)
    VALUES (p_newsletter_id, v_email, p_ip, false, 'rejected_rate_limit');
    RETURN QUERY SELECT false, 429,
      'A confirmation was just sent to this address. Check your inbox or wait a few minutes.'::text,
      NULL::uuid;
    RETURN;
  END IF;

  SELECT count(*) INTO v_email_24h
    FROM newsletter_subscribe_attempts
   WHERE newsletter_id = p_newsletter_id
     AND email = v_email
     AND triggered_email = true
     AND attempted_at > now() - interval '24 hours';
  IF v_email_24h >= 3 THEN
    INSERT INTO newsletter_subscribe_attempts (newsletter_id, email, ip, triggered_email, outcome)
    VALUES (p_newsletter_id, v_email, p_ip, false, 'rejected_rate_limit');
    RETURN QUERY SELECT false, 429,
      'This address has hit today''s confirmation limit for this newsletter. Try again tomorrow.'::text,
      NULL::uuid;
    RETURN;
  END IF;

  INSERT INTO newsletter_subscribe_attempts
    (newsletter_id, email, ip, triggered_email, outcome)
  VALUES
    (p_newsletter_id, v_email, p_ip, p_will_trigger_email, p_outcome)
  RETURNING id INTO v_inserted_id;

  RETURN QUERY SELECT true, 200, NULL::text, v_inserted_id;
END;
$$;

CREATE OR REPLACE FUNCTION log_newsletter_subscribe_rejection(
  p_newsletter_id uuid,
  p_email text,
  p_ip inet,
  p_outcome text,
  p_verification_decision text DEFAULT NULL,
  p_verification_category text DEFAULT NULL,
  p_verification_cached boolean DEFAULT NULL
)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  INSERT INTO newsletter_subscribe_attempts
    (newsletter_id, email, ip, triggered_email, outcome,
     verification_decision, verification_category, verification_cached)
  VALUES
    (p_newsletter_id, lower(trim(p_email)), p_ip, false, p_outcome,
     p_verification_decision, p_verification_category, p_verification_cached);
END;
$$;

-- Session management helpers (Settings -> Sessions tab)
-- Session helpers run as SECURITY DEFINER (need to query auth.sessions
-- which authenticated users can't read directly). Each function enforces
-- a caller-identity check so the elevated privilege can't be abused to
-- read OR terminate other users' sessions. service_role calls bypass
-- the check (auth.role() = 'service_role') because the worker uses them.
--
-- Output hygiene:
--   - `ip` is returned via host(inet) so the /32 CIDR suffix doesn't
--     leak into the UI ("141.98.141.147" not "141.98.141.147/32").
--   - Sessions where the user_agent is a server-side proxy ("node",
--     "Next.js Middleware", "node-fetch", "undici") are filtered out.
--     Each Supabase JWT refresh from the server creates a row here;
--     those aren't real devices and showing them clutters the
--     Sessions UI without giving the user anything actionable.
CREATE OR REPLACE FUNCTION get_user_sessions(p_user_id uuid)
RETURNS TABLE(id uuid, user_agent text, ip text, created_at timestamptz, updated_at timestamptz)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  IF auth.role() <> 'service_role' AND auth.uid() <> p_user_id THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;
  RETURN QUERY
    SELECT
      s.id,
      s.user_agent,
      host(s.ip) AS ip,
      s.created_at,
      s.updated_at
    FROM auth.sessions s
    WHERE s.user_id = p_user_id
      AND (s.not_after IS NULL OR s.not_after > now())
      AND s.user_agent IS NOT NULL
      AND s.user_agent <> ''
      AND s.user_agent <> 'node'
      AND s.user_agent NOT LIKE 'Next.js%'
      AND s.user_agent NOT LIKE 'node-fetch%'
      AND s.user_agent NOT LIKE 'undici%'
    ORDER BY s.updated_at DESC;
END;
$$;

CREATE OR REPLACE FUNCTION get_workspace_sessions(p_workspace_id uuid)
RETURNS TABLE(session_id uuid, user_id uuid, user_email text, full_name text, user_agent text, ip text, created_at timestamptz, updated_at timestamptz)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  -- Workspace members can list sessions for that workspace. Owners +
  -- admins are members too, so the simple membership check covers all
  -- of "anyone with access to this workspace".
  IF auth.role() <> 'service_role'
     AND NOT EXISTS (
       SELECT 1 FROM public.workspace_members m
       WHERE m.workspace_id = p_workspace_id AND m.user_id = auth.uid()
     ) THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;
  RETURN QUERY
    SELECT
      s.id AS session_id,
      s.user_id,
      u.email::text AS user_email,
      COALESCE(p.full_name, u.raw_user_meta_data->>'full_name', u.email)::text AS full_name,
      s.user_agent,
      host(s.ip) AS ip,
      s.created_at,
      s.updated_at
    FROM public.workspace_members wm
    JOIN auth.users u ON u.id = wm.user_id
    JOIN auth.sessions s ON s.user_id = u.id
    LEFT JOIN public.profiles p ON p.id = u.id
    WHERE wm.workspace_id = p_workspace_id
      AND (s.not_after IS NULL OR s.not_after > now())
      AND s.user_agent IS NOT NULL
      AND s.user_agent <> ''
      AND s.user_agent <> 'node'
      AND s.user_agent NOT LIKE 'Next.js%'
      AND s.user_agent NOT LIKE 'node-fetch%'
      AND s.user_agent NOT LIKE 'undici%'
    ORDER BY s.updated_at DESC;
END;
$$;

CREATE OR REPLACE FUNCTION delete_user_session(p_user_id uuid, p_session_id uuid)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  IF auth.role() <> 'service_role' AND auth.uid() <> p_user_id THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;
  DELETE FROM auth.sessions
  WHERE id = p_session_id
    AND user_id = p_user_id;
END;
$$;

-- Lets a workspace OWNER sign out another member's session. delete_user_session
-- only deletes WHERE user_id = caller, so it can't revoke a different user's
-- session — team sign-out from the Sessions tab needs this workspace-scoped one.
CREATE OR REPLACE FUNCTION delete_workspace_session(p_workspace_id uuid, p_session_id uuid)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  -- Only the workspace owner may sign out other members' sessions.
  IF auth.role() <> 'service_role' AND NOT EXISTS (
    SELECT 1 FROM public.workspace_members
    WHERE workspace_id = p_workspace_id AND user_id = auth.uid() AND role = 'owner'
  ) THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;

  -- Only delete sessions belonging to a member of this workspace.
  DELETE FROM auth.sessions s
  USING public.workspace_members wm
  WHERE s.id = p_session_id
    AND s.user_id = wm.user_id
    AND wm.workspace_id = p_workspace_id;
END;
$$;
GRANT EXECUTE ON FUNCTION delete_workspace_session(uuid, uuid) TO authenticated;

-- Email-verification provider key helpers (uses Supabase Vault). Each
-- function enforces a workspace-admin/owner check so the SECURITY
-- DEFINER privilege can't be abused to read or overwrite another
-- workspace's API keys. service_role bypasses the check (worker uses
-- it for first-time setup).
CREATE OR REPLACE FUNCTION set_workspace_verification_key(
  p_workspace_id uuid,
  p_provider text,
  p_api_key text
)
RETURNS uuid
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_old_secret_id uuid;
  v_new_secret_id uuid;
BEGIN
  IF auth.role() <> 'service_role'
     AND NOT EXISTS (
       SELECT 1 FROM workspace_members
       WHERE workspace_id = p_workspace_id
         AND user_id = auth.uid()
         AND role IN ('owner', 'admin')
     ) THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;
  IF p_provider NOT IN ('native') THEN
    RAISE EXCEPTION 'Unsupported verification provider: %', p_provider;
  END IF;
  IF p_api_key IS NULL OR length(trim(p_api_key)) = 0 THEN
    RAISE EXCEPTION 'API key cannot be empty';
  END IF;

  SELECT email_verification_secret_id INTO v_old_secret_id
    FROM workspaces WHERE id = p_workspace_id;

  v_new_secret_id := vault.create_secret(
    p_api_key,
    NULL,
    format('email-verification:%s:%s', p_provider, p_workspace_id::text)
  );

  UPDATE workspaces
     SET email_verification_provider = p_provider,
         email_verification_secret_id = v_new_secret_id
   WHERE id = p_workspace_id;

  IF v_old_secret_id IS NOT NULL THEN
    DELETE FROM vault.secrets WHERE id = v_old_secret_id;
  END IF;

  RETURN v_new_secret_id;
END;
$$;

CREATE OR REPLACE FUNCTION get_workspace_verification_key(p_workspace_id uuid)
RETURNS TABLE(provider text, api_key text)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_provider text;
  v_secret_id uuid;
  v_decrypted text;
BEGIN
  -- Admin/owner-only — the decrypted API key is what a third-party
  -- verification provider charges money for, and we don't want
  -- non-admin members able to read it.
  IF auth.role() <> 'service_role'
     AND NOT EXISTS (
       SELECT 1 FROM workspace_members
       WHERE workspace_id = p_workspace_id
         AND user_id = auth.uid()
         AND role IN ('owner', 'admin')
     ) THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;

  SELECT email_verification_provider, email_verification_secret_id
    INTO v_provider, v_secret_id
    FROM workspaces WHERE id = p_workspace_id;

  IF v_provider IS NULL OR v_secret_id IS NULL THEN
    RETURN;
  END IF;

  SELECT decrypted_secret INTO v_decrypted
    FROM vault.decrypted_secrets WHERE id = v_secret_id;

  IF v_decrypted IS NULL THEN
    RETURN;
  END IF;

  RETURN QUERY SELECT v_provider, v_decrypted;
END;
$$;

CREATE OR REPLACE FUNCTION clear_workspace_verification_key(p_workspace_id uuid)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_old_secret_id uuid;
BEGIN
  IF auth.role() <> 'service_role'
     AND NOT EXISTS (
       SELECT 1 FROM workspace_members
       WHERE workspace_id = p_workspace_id
         AND user_id = auth.uid()
         AND role IN ('owner', 'admin')
     ) THEN
    RAISE EXCEPTION 'Forbidden' USING ERRCODE = 'insufficient_privilege';
  END IF;

  SELECT email_verification_secret_id INTO v_old_secret_id
    FROM workspaces WHERE id = p_workspace_id;

  UPDATE workspaces
     SET email_verification_provider = NULL,
         email_verification_secret_id = NULL
   WHERE id = p_workspace_id;

  IF v_old_secret_id IS NOT NULL THEN
    DELETE FROM vault.secrets WHERE id = v_old_secret_id;
  END IF;
END;
$$;

-- ---- inbound_domain stats helpers --------------------------------
-- Two RPCs the worker + smtp-server invoke when mail arrives on a
-- legacy inbound_domains row. Both bump the per-domain counters; the
-- "_on_email" variant also auto-promotes a still-pending domain to
-- 'verified' on first receipt and reports back so the caller can log
-- the transition. service_role-only — worker uses it; no end-user
-- callable surface.
CREATE OR REPLACE FUNCTION public.update_inbound_domain_on_email(p_domain_id uuid)
RETURNS TABLE(was_pending boolean)
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
DECLARE
  v_was_pending boolean;
BEGIN
  SELECT verification_status = 'pending'
    INTO v_was_pending
    FROM inbound_domains
   WHERE id = p_domain_id
   FOR UPDATE;

  IF NOT FOUND THEN
    RETURN; -- row disappeared between INSERT and stats update; no-op
  END IF;

  UPDATE inbound_domains
     SET emails_received_count = COALESCE(emails_received_count, 0) + 1,
         last_email_at         = now(),
         verification_status   = CASE WHEN v_was_pending THEN 'verified' ELSE verification_status END,
         updated_at            = now()
   WHERE id = p_domain_id;

  RETURN QUERY SELECT v_was_pending;
END;
$$;

REVOKE EXECUTE ON FUNCTION public.update_inbound_domain_on_email(uuid)
  FROM PUBLIC, anon, authenticated;
GRANT  EXECUTE ON FUNCTION public.update_inbound_domain_on_email(uuid)
  TO service_role;

CREATE OR REPLACE FUNCTION public.increment_inbound_domain_stats(domain_id uuid)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  UPDATE inbound_domains
     SET emails_received_count = COALESCE(emails_received_count, 0) + 1,
         last_email_at         = now(),
         updated_at            = now()
   WHERE id = domain_id;
END;
$$;

REVOKE EXECUTE ON FUNCTION public.increment_inbound_domain_stats(uuid)
  FROM PUBLIC, anon, authenticated;
GRANT  EXECUTE ON FUNCTION public.increment_inbound_domain_stats(uuid)
  TO service_role;


-- =====================================================================
-- SECTION: row level security (RLS)
-- =====================================================================

-- Enable RLS everywhere
ALTER TABLE profiles ENABLE ROW LEVEL SECURITY;
ALTER TABLE workspaces ENABLE ROW LEVEL SECURITY;
ALTER TABLE workspace_members ENABLE ROW LEVEL SECURITY;
ALTER TABLE workspace_invites ENABLE ROW LEVEL SECURITY;
ALTER TABLE api_keys ENABLE ROW LEVEL SECURITY;
ALTER TABLE email_logs ENABLE ROW LEVEL SECURITY;
ALTER TABLE email_events ENABLE ROW LEVEL SECURITY;
ALTER TABLE email_templates ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduled_emails ENABLE ROW LEVEL SECURITY;
ALTER TABLE rate_limits ENABLE ROW LEVEL SECURITY;
ALTER TABLE domain_health ENABLE ROW LEVEL SECURITY;
ALTER TABLE reputation_metrics ENABLE ROW LEVEL SECURITY;
ALTER TABLE inbound_domains ENABLE ROW LEVEL SECURITY;
ALTER TABLE inbound_emails ENABLE ROW LEVEL SECURITY;
ALTER TABLE template_images ENABLE ROW LEVEL SECURITY;
ALTER TABLE api_request_logs ENABLE ROW LEVEL SECURITY;
ALTER TABLE contacts ENABLE ROW LEVEL SECURITY;
ALTER TABLE contact_imports ENABLE ROW LEVEL SECURITY;
ALTER TABLE suppression_list ENABLE ROW LEVEL SECURITY;
ALTER TABLE soft_bounce_tracker ENABLE ROW LEVEL SECURITY;
ALTER TABLE mcp_auth_codes ENABLE ROW LEVEL SECURITY;
ALTER TABLE mcp_oauth_clients ENABLE ROW LEVEL SECURITY;
ALTER TABLE mfa_backup_codes ENABLE ROW LEVEL SECURITY;
ALTER TABLE email_verification_cache ENABLE ROW LEVEL SECURITY;
ALTER TABLE broadcasts ENABLE ROW LEVEL SECURITY;
ALTER TABLE newsletters ENABLE ROW LEVEL SECURITY;
ALTER TABLE newsletter_issues ENABLE ROW LEVEL SECURITY;
ALTER TABLE newsletter_subscribers ENABLE ROW LEVEL SECURITY;
ALTER TABLE newsletter_subscribe_attempts ENABLE ROW LEVEL SECURITY;
ALTER TABLE automations ENABLE ROW LEVEL SECURITY;
ALTER TABLE automation_steps ENABLE ROW LEVEL SECURITY;
ALTER TABLE automation_enrollments ENABLE ROW LEVEL SECURITY;
ALTER TABLE automation_events ENABLE ROW LEVEL SECURITY;
ALTER TABLE automation_test_runs ENABLE ROW LEVEL SECURITY;
ALTER TABLE webhook_endpoints ENABLE ROW LEVEL SECURITY;
ALTER TABLE webhook_deliveries ENABLE ROW LEVEL SECURITY;


-- ---- profiles ------------------------------------------------------
CREATE POLICY "Users can insert own profile" ON profiles FOR INSERT WITH CHECK (auth.uid() = id);
CREATE POLICY "Users can update own profile" ON profiles FOR UPDATE USING (auth.uid() = id);
CREATE POLICY "Workspace members can view co-member profiles" ON profiles FOR SELECT
  USING (
    (auth.uid() = id) OR
    (EXISTS (
      SELECT 1 FROM workspace_members wm1
      JOIN workspace_members wm2 ON wm1.workspace_id = wm2.workspace_id
      WHERE wm1.user_id = auth.uid() AND wm2.user_id = profiles.id
    ))
  );

-- ---- workspaces ----------------------------------------------------
CREATE POLICY "Users can create workspaces" ON workspaces FOR INSERT WITH CHECK (auth.uid() = created_by);
CREATE POLICY "Members can view their workspaces" ON workspaces FOR SELECT
  USING ((created_by = auth.uid()) OR is_workspace_member(id, auth.uid()));
CREATE POLICY "Owners and admins can update workspaces" ON workspaces FOR UPDATE
  USING (is_workspace_admin_or_owner(id, auth.uid()));
CREATE POLICY "Owners can delete workspaces" ON workspaces FOR DELETE
  USING (EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = workspaces.id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role = 'owner'
  ));

-- ---- workspace_members ---------------------------------------------
CREATE POLICY "Members can view workspace members" ON workspace_members FOR SELECT
  USING (is_workspace_member(workspace_id, auth.uid()));
CREATE POLICY "Owners and admins can add members" ON workspace_members FOR INSERT
  WITH CHECK (
    ((user_id = auth.uid()) AND (role = 'owner')) OR
    is_workspace_admin_or_owner(workspace_id, auth.uid())
  );
CREATE POLICY "Owners and admins can update members" ON workspace_members FOR UPDATE
  USING (is_workspace_admin_or_owner(workspace_id, auth.uid()));
CREATE POLICY "Owners admins can remove members or self-remove" ON workspace_members FOR DELETE
  USING ((user_id = auth.uid()) OR is_workspace_admin_or_owner(workspace_id, auth.uid()));

-- ---- workspace_invites ---------------------------------------------
CREATE POLICY "Workspace members can view invites" ON workspace_invites FOR SELECT
  USING (is_workspace_member(workspace_id, auth.uid()));
CREATE POLICY "Admins and owners can create invites" ON workspace_invites FOR INSERT
  WITH CHECK (is_workspace_admin_or_owner(workspace_id, auth.uid()));
CREATE POLICY "Admins and owners can update invites" ON workspace_invites FOR UPDATE
  USING (is_workspace_admin_or_owner(workspace_id, auth.uid()));
CREATE POLICY "Invitees can accept their own invites" ON workspace_invites FOR UPDATE
  USING (email = (auth.jwt() ->> 'email'));
CREATE POLICY "Owners and admins can delete invites" ON workspace_invites FOR DELETE
  USING (is_workspace_admin_or_owner(workspace_id, auth.uid()));

-- ---- api_keys ------------------------------------------------------
-- Service role bypasses RLS; the worker uses SUPABASE_SERVICE_ROLE_KEY
-- to verify API keys. Dashboard CRUD goes through the owner policy.
CREATE POLICY "Users can manage own API keys" ON api_keys FOR ALL USING (auth.uid() = user_id);

-- ---- email_logs (workspace-scoped) ---------------------------------
CREATE POLICY "Workspace members can view email logs" ON email_logs
  FOR SELECT TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  );
-- email_logs is only inserted by the worker, which uses the service-role
-- key (service_role bypasses RLS via the policy below). We DO NOT grant
-- INSERT to anon/authenticated — that would let any signed-in user
-- fabricate delivery logs (forge "sent" records, hide bounces, etc.).
CREATE POLICY "Service role can manage email logs" ON email_logs
  FOR ALL USING (auth.role() = 'service_role');

-- ---- email_events --------------------------------------------------
CREATE POLICY "Workspace members can view email events" ON email_events
  FOR SELECT TO authenticated
  USING (
    EXISTS (
      SELECT 1 FROM email_logs
      WHERE email_logs.id = email_events.email_log_id
        AND (
          (email_logs.workspace_id IS NULL AND email_logs.user_id = auth.uid())
          OR email_logs.workspace_id IN (
            SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
          )
        )
    )
  );
CREATE POLICY "Service role can manage email events" ON email_events
  FOR ALL USING (auth.role() = 'service_role');

-- ---- email_templates -----------------------------------------------
CREATE POLICY "Workspace members can manage email templates" ON email_templates
  FOR ALL TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  )
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  );

-- ---- scheduled_emails ----------------------------------------------
CREATE POLICY "Workspace members can view scheduled emails" ON scheduled_emails
  FOR SELECT TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  );
CREATE POLICY "Workspace members can insert scheduled emails" ON scheduled_emails
  FOR INSERT TO authenticated
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  );
CREATE POLICY "Workspace members can update scheduled emails" ON scheduled_emails
  FOR UPDATE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  );
CREATE POLICY "Workspace members can delete scheduled emails" ON scheduled_emails
  FOR DELETE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (
      SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()
    )
  );

-- ---- rate_limits ---------------------------------------------------
CREATE POLICY "Users can view own rate limits" ON rate_limits FOR SELECT USING (auth.uid() = user_id);
CREATE POLICY "Users can insert own rate limits" ON rate_limits FOR INSERT WITH CHECK (auth.uid() = user_id);
CREATE POLICY "Users can update own rate limits" ON rate_limits FOR UPDATE USING (auth.uid() = user_id);
CREATE POLICY "Service role can manage rate limits" ON rate_limits FOR ALL USING (auth.role() = 'service_role');

-- ---- domain_health (workspace-scoped) ------------------------------
CREATE POLICY "Workspace members can view domain health" ON domain_health
  FOR SELECT TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can insert domain health" ON domain_health
  FOR INSERT TO authenticated
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can update domain health" ON domain_health
  FOR UPDATE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can delete domain health" ON domain_health
  FOR DELETE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- reputation_metrics --------------------------------------------
CREATE POLICY "Workspace members can view reputation metrics" ON reputation_metrics
  FOR SELECT TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can insert reputation metrics" ON reputation_metrics
  FOR INSERT TO authenticated
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can update reputation metrics" ON reputation_metrics
  FOR UPDATE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- inbound_domains -----------------------------------------------
CREATE POLICY "Workspace members can view inbound domains" ON inbound_domains
  FOR SELECT TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can insert inbound domains" ON inbound_domains
  FOR INSERT TO authenticated
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can update inbound domains" ON inbound_domains
  FOR UPDATE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can delete inbound domains" ON inbound_domains
  FOR DELETE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- inbound_emails ------------------------------------------------
CREATE POLICY "Workspace members can view inbound emails" ON inbound_emails
  FOR SELECT TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can insert inbound emails" ON inbound_emails
  FOR INSERT TO authenticated
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can update inbound emails" ON inbound_emails
  FOR UPDATE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members can delete inbound emails" ON inbound_emails
  FOR DELETE TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- template_images -----------------------------------------------
CREATE POLICY "Users can manage own workspace template images" ON template_images
  FOR ALL USING (
    workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- api_request_logs ----------------------------------------------
CREATE POLICY "Users can view workspace request logs" ON api_request_logs
  FOR SELECT TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- ---- contacts ------------------------------------------------------
CREATE POLICY "Users can manage workspace contacts" ON contacts
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- ---- contact_removals (removed-email history) ----------------------
-- Members read + clear history; inserts come from the web (manual) or the
-- worker (service_role, bypasses RLS, for automatic bounce/complaint removals).
ALTER TABLE contact_removals ENABLE ROW LEVEL SECURITY;
CREATE POLICY "Workspace members view contact removals" ON contact_removals
  FOR SELECT TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));
CREATE POLICY "Workspace members create contact removals" ON contact_removals
  FOR INSERT TO authenticated
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));
CREATE POLICY "Workspace members delete contact removals" ON contact_removals
  FOR DELETE TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- ---- contact_imports -----------------------------------------------
-- Members can see imports in their workspace + create new ones tied
-- to their own user_id. Worker (service_role) bypasses RLS so the
-- progress / completion writes don't need an authenticated policy.
CREATE POLICY "Workspace members view contact imports" ON contact_imports
  FOR SELECT TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));
CREATE POLICY "Workspace members create contact imports" ON contact_imports
  FOR INSERT TO authenticated
  WITH CHECK (
    user_id = auth.uid()
    AND workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );
CREATE POLICY "Workspace members cancel own imports" ON contact_imports
  FOR UPDATE TO authenticated
  USING (
    user_id = auth.uid()
    AND workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- suppression_list ----------------------------------------------
CREATE POLICY "Users can view workspace suppressions" ON suppression_list
  FOR SELECT TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));
CREATE POLICY "Users can insert workspace suppressions" ON suppression_list
  FOR INSERT TO authenticated
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));
CREATE POLICY "Users can delete workspace suppressions" ON suppression_list
  FOR DELETE TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- ---- soft_bounce_tracker (service-role only; worker writes via SECURITY DEFINER funcs) ----
-- No authenticated policies — surfaced to the dashboard only via aggregates.

-- ---- mcp_auth_codes ------------------------------------------------
CREATE POLICY "Users can view own mcp_auth_codes" ON mcp_auth_codes
  FOR SELECT USING (auth.uid() = user_id);
CREATE POLICY "Users can insert own mcp_auth_codes" ON mcp_auth_codes
  FOR INSERT WITH CHECK (auth.uid() = user_id);
CREATE POLICY "Users can update own mcp_auth_codes" ON mcp_auth_codes
  FOR UPDATE USING (auth.uid() = user_id);
CREATE POLICY "Users can delete own mcp_auth_codes" ON mcp_auth_codes
  FOR DELETE USING (auth.uid() = user_id);
CREATE POLICY "Service role full access" ON mcp_auth_codes
  FOR ALL TO service_role USING (true) WITH CHECK (true);

-- ---- mcp_oauth_clients (service-role only) -------------------------
-- Worker uses SUPABASE_SERVICE_ROLE_KEY; no anon/authenticated policies.

-- ---- mfa_backup_codes ----------------------------------------------
CREATE POLICY "Users manage own MFA backup codes" ON mfa_backup_codes
  FOR ALL USING (user_id = auth.uid()) WITH CHECK (user_id = auth.uid());

-- ---- email_verification_cache --------------------------------------
CREATE POLICY "Workspace members can view verification cache" ON email_verification_cache
  FOR SELECT TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- ---- broadcasts ----------------------------------------------------
CREATE POLICY "Workspace members can manage broadcasts" ON broadcasts
  FOR ALL TO authenticated
  USING (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  )
  WITH CHECK (
    workspace_id IS NULL AND auth.uid() = user_id
    OR workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
  );

-- ---- newsletters ---------------------------------------------------
CREATE POLICY "Workspace members can manage newsletters" ON newsletters
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

CREATE POLICY "Workspace members can manage newsletter issues" ON newsletter_issues
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

CREATE POLICY "Workspace members can manage newsletter subscribers" ON newsletter_subscribers
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- The public subscribe endpoint hits this table via a SECURITY DEFINER
-- RPC (check_and_log_newsletter_subscribe), so we don't need anon policies
-- here. Service role bypasses RLS for the worker.

-- newsletter_subscribe_attempts: service-role only (no end-user access)

-- ---- automations ---------------------------------------------------
CREATE POLICY "Workspace members can manage automations" ON automations
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

CREATE POLICY "Workspace members can manage automation steps" ON automation_steps
  FOR ALL TO authenticated
  USING (
    automation_id IN (
      SELECT id FROM automations
      WHERE workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
    )
  )
  WITH CHECK (
    automation_id IN (
      SELECT id FROM automations
      WHERE workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
    )
  );

CREATE POLICY "Workspace members can manage enrollments" ON automation_enrollments
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

CREATE POLICY "Workspace members can view automation events" ON automation_events
  FOR SELECT TO authenticated
  USING (
    enrollment_id IN (
      SELECT id FROM automation_enrollments
      WHERE workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid())
    )
  );

CREATE POLICY "Workspace members can manage automation test runs" ON automation_test_runs
  FOR ALL TO authenticated
  USING (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()))
  WITH CHECK (workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = auth.uid()));

-- ---- webhook_endpoints + webhook_deliveries ------------------------
CREATE POLICY "Workspace members can view webhook endpoints" ON webhook_endpoints
  FOR SELECT
  USING (EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = webhook_endpoints.workspace_id
      AND workspace_members.user_id = auth.uid()
  ));
CREATE POLICY "Workspace admins and owners can insert webhook endpoints" ON webhook_endpoints
  FOR INSERT
  WITH CHECK (EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = webhook_endpoints.workspace_id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role IN ('owner', 'admin')
  ));
CREATE POLICY "Workspace admins and owners can update webhook endpoints" ON webhook_endpoints
  FOR UPDATE
  USING (EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = webhook_endpoints.workspace_id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role IN ('owner', 'admin')
  ));
CREATE POLICY "Workspace admins and owners can delete webhook endpoints" ON webhook_endpoints
  FOR DELETE
  USING (EXISTS (
    SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = webhook_endpoints.workspace_id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role IN ('owner', 'admin')
  ));

CREATE POLICY "Workspace members can view webhook deliveries" ON webhook_deliveries
  FOR SELECT
  USING (EXISTS (
    SELECT 1 FROM webhook_endpoints
    JOIN workspace_members ON workspace_members.workspace_id = webhook_endpoints.workspace_id
    WHERE webhook_endpoints.id = webhook_deliveries.webhook_endpoint_id
      AND workspace_members.user_id = auth.uid()
  ));


-- =====================================================================
-- SECTION: storage buckets
-- =====================================================================
-- Four buckets are required for the dashboard to work end-to-end:
--
--   avatars         (public)  — user profile photos + workspace logos
--                               under {profiles,workspaces}/{id}/
--   template-images (public)  — email-template image assets
--                               under {workspace_id}/{template_id}/
--   contact-imports (private) — CSV uploads pending worker processing
--                               under {workspace_id}/{import_id}.csv
--   exports         (private) — generated request-log CSV exports the
--                               worker writes (service_role) and hands to
--                               the user via a short-lived signed URL
--
-- Without these the avatar/logo/template/import/export flows all error
-- with "Bucket not found". Bucket rows live in the `storage` schema
-- managed by Supabase — INSERT is allowed at install time.

INSERT INTO storage.buckets (id, name, public, file_size_limit, allowed_mime_types) VALUES
  ('avatars',         'avatars',         true,  5242880,
    ARRAY['image/png','image/jpeg','image/jpg','image/webp','image/gif']),
  ('template-images', 'template-images', true,  5242880,
    ARRAY['image/png','image/jpeg','image/jpg','image/webp','image/gif']),
  ('contact-imports', 'contact-imports', false, 20971520,
    ARRAY['text/csv','application/vnd.ms-excel','application/octet-stream']),
  -- request-log CSV exports: worker-written, downloaded via signed URL.
  -- Unrestricted mime (worker sets text/csv); 100 MB ceiling for large logs.
  ('exports',         'exports',         false, 104857600, NULL)
ON CONFLICT (id) DO UPDATE SET
  public             = EXCLUDED.public,
  file_size_limit    = EXCLUDED.file_size_limit,
  allowed_mime_types = EXCLUDED.allowed_mime_types;

-- Policies on storage.objects. Avatars + template-images are world-
-- readable so <img> tags work in emails / public profile pages.
-- Writes scoped to the caller's user_id or workspace membership.
-- contact-imports is fully private; only service_role (worker) reads.

CREATE POLICY "Avatars public read" ON storage.objects FOR SELECT
  USING (bucket_id = 'avatars');

CREATE POLICY "Avatars authenticated upload" ON storage.objects FOR INSERT
  TO authenticated
  WITH CHECK (
    bucket_id = 'avatars'
    AND (
      ((storage.foldername(name))[1] = 'profiles'
        AND (storage.foldername(name))[2] = auth.uid()::text)
      OR
      ((storage.foldername(name))[1] = 'workspaces'
        AND EXISTS (
          SELECT 1 FROM public.workspace_members
          WHERE workspace_id::text = (storage.foldername(name))[2]
            AND user_id = auth.uid()
        ))
    )
  );

CREATE POLICY "Avatars authenticated update" ON storage.objects FOR UPDATE
  TO authenticated
  USING (
    bucket_id = 'avatars'
    AND (
      ((storage.foldername(name))[1] = 'profiles'
        AND (storage.foldername(name))[2] = auth.uid()::text)
      OR
      ((storage.foldername(name))[1] = 'workspaces'
        AND EXISTS (
          SELECT 1 FROM public.workspace_members
          WHERE workspace_id::text = (storage.foldername(name))[2]
            AND user_id = auth.uid()
        ))
    )
  );

CREATE POLICY "Avatars authenticated delete" ON storage.objects FOR DELETE
  TO authenticated
  USING (
    bucket_id = 'avatars'
    AND (
      ((storage.foldername(name))[1] = 'profiles'
        AND (storage.foldername(name))[2] = auth.uid()::text)
      OR
      ((storage.foldername(name))[1] = 'workspaces'
        AND EXISTS (
          SELECT 1 FROM public.workspace_members
          WHERE workspace_id::text = (storage.foldername(name))[2]
            AND user_id = auth.uid()
        ))
    )
  );

CREATE POLICY "Template images public read" ON storage.objects FOR SELECT
  USING (bucket_id = 'template-images');

CREATE POLICY "Template images workspace upload" ON storage.objects FOR INSERT
  TO authenticated
  WITH CHECK (
    bucket_id = 'template-images'
    AND EXISTS (
      SELECT 1 FROM public.workspace_members
      WHERE workspace_id::text = (storage.foldername(name))[1]
        AND user_id = auth.uid()
    )
  );

CREATE POLICY "Template images workspace delete" ON storage.objects FOR DELETE
  TO authenticated
  USING (
    bucket_id = 'template-images'
    AND EXISTS (
      SELECT 1 FROM public.workspace_members
      WHERE workspace_id::text = (storage.foldername(name))[1]
        AND user_id = auth.uid()
    )
  );

-- contact-imports is upload-only for end users; reads happen via
-- service_role inside the worker.
CREATE POLICY "Contact imports workspace upload" ON storage.objects FOR INSERT
  TO authenticated
  WITH CHECK (
    bucket_id = 'contact-imports'
    AND EXISTS (
      SELECT 1 FROM public.workspace_members
      WHERE workspace_id::text = (storage.foldername(name))[1]
        AND user_id = auth.uid()
    )
  );

-- exports: intentionally NO authenticated storage policy. The worker writes
-- the CSV with the service_role key (bypasses RLS) and the user downloads it
-- through a short-lived signed URL (also bypasses RLS), so end users never
-- touch storage.objects for this bucket directly.


-- =====================================================================
-- Done.
-- =====================================================================
-- SECTION: cloud_domains (unified Domains UI — both editions)
-- =====================================================================
-- Originally cloud-only, lifted into self-host so both editions use the
-- same /dashboard/domains UI for sending + receiving setup. The worker
-- skips VDM tenants / multi-region / configuration sets when
-- EDITION=self-hosted; the cloud-only columns (tenant_name,
-- configuration_set_name) stay nullable here for forward-compat.
-- =====================================================================

CREATE TABLE IF NOT EXISTS cloud_domains (
  id                   uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  workspace_id         uuid NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
  domain               text NOT NULL,
  ses_region           text NOT NULL DEFAULT 'us-east-1',
  ses_identity_status  text NOT NULL DEFAULT 'NOT_STARTED'
    CHECK (ses_identity_status IN ('NOT_STARTED', 'PENDING', 'SUCCESS', 'FAILED', 'TEMPORARY_FAILURE')),
  dkim_tokens          text[] NOT NULL DEFAULT '{}',
  verification_token   text,
  mail_from_domain     text,
  mail_from_mx_record  text,
  mail_from_txt_record text,
  dkim_selector        text,
  dkim_public_key      text,
  dkim_verified        boolean NOT NULL DEFAULT false,
  spf_verified         boolean NOT NULL DEFAULT false,
  sending_enabled      boolean NOT NULL DEFAULT false,
  receiving_enabled    boolean NOT NULL DEFAULT false,
  -- Whether the receiving-region identity is verified (separate from sending).
  receiving_verified   boolean NOT NULL DEFAULT false,
  receipt_rule_name    text,
  webhook_url          text,
  webhook_secret       text,
  tenant_name          text,
  configuration_set_name text,
  max_daily_emails     integer,
  max_monthly_emails   integer,
  verification_attempts int NOT NULL DEFAULT 0,
  last_checked_at      timestamptz,
  created_at           timestamptz NOT NULL DEFAULT now(),
  updated_at           timestamptz NOT NULL DEFAULT now(),
  -- Global uniqueness — a domain can only belong to one workspace at
  -- a time. Was previously UNIQUE(workspace_id, domain) which allowed
  -- the same domain in multiple workspaces; SES BYODKIM silently broke
  -- the first workspace when the second added the same domain.
  CONSTRAINT cloud_domains_domain_unique UNIQUE (domain)
);

CREATE INDEX IF NOT EXISTS idx_cloud_domains_workspace_id ON cloud_domains(workspace_id);
CREATE INDEX IF NOT EXISTS idx_cloud_domains_domain ON cloud_domains(domain);
CREATE INDEX IF NOT EXISTS idx_cloud_domains_ses_status ON cloud_domains(ses_identity_status);

CREATE TRIGGER update_cloud_domains_updated_at
  BEFORE UPDATE ON cloud_domains
  FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();

ALTER TABLE cloud_domains ENABLE ROW LEVEL SECURITY;

CREATE POLICY "Workspace members can view their cloud domains"
  ON cloud_domains FOR SELECT
  USING (EXISTS (SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = cloud_domains.workspace_id
      AND workspace_members.user_id = auth.uid()));

CREATE POLICY "Workspace admins and owners can insert cloud domains"
  ON cloud_domains FOR INSERT
  WITH CHECK (EXISTS (SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = cloud_domains.workspace_id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role IN ('owner', 'admin')));

CREATE POLICY "Workspace admins and owners can update cloud domains"
  ON cloud_domains FOR UPDATE
  USING (EXISTS (SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = cloud_domains.workspace_id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role IN ('owner', 'admin')));

CREATE POLICY "Workspace admins and owners can delete cloud domains"
  ON cloud_domains FOR DELETE
  USING (EXISTS (SELECT 1 FROM workspace_members
    WHERE workspace_members.workspace_id = cloud_domains.workspace_id
      AND workspace_members.user_id = auth.uid()
      AND workspace_members.role IN ('owner', 'admin')));

-- BYODKIM private keys per cloud domain. Reused for the receiving-region
-- identity so it verifies against the same posthawk._domainkey record already
-- in DNS. RLS on with NO policies — service-role (worker) only; the key must
-- never reach the browser (so not a column on cloud_domains / select('*')).
CREATE TABLE IF NOT EXISTS cloud_domain_dkim_keys (
  cloud_domain_id uuid PRIMARY KEY REFERENCES cloud_domains(id) ON DELETE CASCADE,
  private_key     text NOT NULL,
  created_at      timestamptz NOT NULL DEFAULT now()
);

ALTER TABLE cloud_domain_dkim_keys ENABLE ROW LEVEL SECURITY;
-- (intentionally no policies — service-role-only access)

-- BLOCK direct user-supplied flips of `sending_enabled = true` unless
-- the SES identity is actually verified. Without this, a workspace
-- admin could `UPDATE cloud_domains SET sending_enabled = true` from
-- the dashboard's Supabase client and bypass:
--   - The invite-team-member gate
--   - The send-time domain check in EmailService
-- both of which read this column. The worker's CloudDomainService
-- (service_role) is the only path that should be flipping this — it
-- only does so after `ses_identity_status='SUCCESS'`.
CREATE OR REPLACE FUNCTION enforce_cloud_domain_sending_gate()
RETURNS TRIGGER
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
  -- service_role can do anything (worker driving the verification flow).
  IF auth.role() = 'service_role' THEN
    RETURN NEW;
  END IF;

  -- User attempted to turn sending on. Only allow if SES has already
  -- confirmed the identity is verified.
  IF NEW.sending_enabled = true
     AND (OLD.sending_enabled IS DISTINCT FROM true)
     AND NEW.ses_identity_status <> 'SUCCESS' THEN
    RAISE EXCEPTION
      'Cannot enable sending until SES verifies the domain (ses_identity_status = %). The worker flips this automatically once DKIM/SPF clear.', NEW.ses_identity_status
      USING ERRCODE = 'check_violation';
  END IF;

  -- Defense in depth: prevent a user from also forging ses_identity_status
  -- itself. Only service_role can change that field.
  IF NEW.ses_identity_status IS DISTINCT FROM OLD.ses_identity_status THEN
    RAISE EXCEPTION
      'ses_identity_status is read-only via user JWT — only the worker updates this'
      USING ERRCODE = 'insufficient_privilege';
  END IF;

  RETURN NEW;
END;
$$;

DROP TRIGGER IF EXISTS cloud_domains_sending_gate ON cloud_domains;
CREATE TRIGGER cloud_domains_sending_gate
  BEFORE UPDATE ON cloud_domains
  FOR EACH ROW EXECUTE FUNCTION enforce_cloud_domain_sending_gate();

-- TODO(security): tighten webhook_secret visibility — the SELECT
-- policy above lets all workspace members read the row including the
-- inbound webhook HMAC. A non-admin member could use it to forge
-- inbound webhook payloads to the customer's webhook URL. The clean
-- fix requires either a column-level GRANT + dashboard refactor to
-- fetch the secret via a service-role API route, or splitting the
-- secret into a separate admin-only table. Tracked separately.

-- Link inbound_emails to cloud_domains so the unified Domains UI shows
-- inbound mail received per domain. inbound_domain_id stays nullable.
ALTER TABLE inbound_emails
  ADD COLUMN IF NOT EXISTS cloud_domain_id uuid REFERENCES cloud_domains(id) ON DELETE SET NULL;

CREATE INDEX IF NOT EXISTS idx_inbound_emails_cloud_domain_id
  ON inbound_emails(cloud_domain_id);

-- =====================================================================
-- Verify with:
--   SELECT count(*) FROM information_schema.tables
--    WHERE table_schema = 'public';
-- Expect ~36 tables.

-- ===========================================
-- RLS PERFORMANCE: wrap auth.*() in (select ...)
-- ===========================================
-- Evaluate auth.uid()/jwt()/role() once per query (initPlan) instead of once
-- per row — 100x+ on large tables. Idempotent; runs after all policies are
-- created. Safe to re-run.
DO $$
DECLARE r record; nq text; nc text; changed boolean;
BEGIN
  FOR r IN
    SELECT schemaname, tablename, policyname, qual, with_check
    FROM pg_policies
    WHERE schemaname = 'public'
      AND ((qual IS NOT NULL AND qual ~ 'auth\.(uid|jwt|role)\(\)' AND qual !~* 'select auth\.')
        OR (with_check IS NOT NULL AND with_check ~ 'auth\.(uid|jwt|role)\(\)' AND with_check !~* 'select auth\.'))
  LOOP
    nq := r.qual; nc := r.with_check; changed := false;
    IF r.qual IS NOT NULL AND r.qual ~ 'auth\.(uid|jwt|role)\(\)' AND r.qual !~* 'select auth\.' THEN
      nq := regexp_replace(r.qual, 'auth\.(uid|jwt|role)\(\)', '(select auth.\1())', 'g'); changed := true;
    END IF;
    IF r.with_check IS NOT NULL AND r.with_check ~ 'auth\.(uid|jwt|role)\(\)' AND r.with_check !~* 'select auth\.' THEN
      nc := regexp_replace(r.with_check, 'auth\.(uid|jwt|role)\(\)', '(select auth.\1())', 'g'); changed := true;
    END IF;
    IF NOT changed THEN CONTINUE; END IF;
    IF nq IS NOT NULL AND nc IS NOT NULL THEN
      EXECUTE format('ALTER POLICY %I ON %I.%I USING (%s) WITH CHECK (%s)', r.policyname, r.schemaname, r.tablename, nq, nc);
    ELSIF nq IS NOT NULL THEN
      EXECUTE format('ALTER POLICY %I ON %I.%I USING (%s)', r.policyname, r.schemaname, r.tablename, nq);
    ELSE
      EXECUTE format('ALTER POLICY %I ON %I.%I WITH CHECK (%s)', r.policyname, r.schemaname, r.tablename, nc);
    END IF;
  END LOOP;
END $$;

-- ===========================================
-- EMAIL STATS DAILY + DASHBOARD STATS RPC
-- ===========================================
-- email_stats_daily exists in both editions so get_dashboard_stats (used by
-- /api/stats) can reference it. Self-host has no retention, so this table
-- stays empty and the archive sums to 0 — live email_logs cover everything.
create table if not exists public.email_stats_daily (
  workspace_id uuid not null references public.workspaces(id) on delete cascade,
  day date not null,
  total integer not null default 0,
  delivered integer not null default 0,
  hard_bounced integer not null default 0,
  soft_bounced integer not null default 0,
  complained integer not null default 0,
  failed integer not null default 0,
  opened integer not null default 0,
  updated_at timestamptz not null default now(),
  primary key (workspace_id, day)
);
alter table public.email_stats_daily enable row level security;
drop policy if exists "members read workspace email stats" on public.email_stats_daily;
create policy "members read workspace email stats"
  on public.email_stats_daily for select
  using (
    exists (select 1 from public.workspace_members m
            where m.workspace_id = email_stats_daily.workspace_id and m.user_id = (select auth.uid()))
    or exists (select 1 from public.workspaces w
            where w.id = email_stats_daily.workspace_id and w.created_by = (select auth.uid()))
  );

create or replace function public.get_dashboard_stats(p_workspace_id uuid)
returns table(
  emails_sent bigint, delivered bigint, active_keys bigint,
  emails_today bigint, emails_yesterday bigint, delivered_today bigint, delivered_yesterday bigint
)
language plpgsql stable security definer set search_path to 'public'
as $$
declare uid uuid := auth.uid(); allowed boolean; d_set text[] := array['sent','delivered','opened','clicked'];
begin
  if p_workspace_id is null then allowed := (uid is not null);
  else allowed := exists (select 1 from workspace_members where workspace_id = p_workspace_id and user_id = uid)
              or exists (select 1 from workspaces where id = p_workspace_id and created_by = uid);
  end if;
  if not allowed then
    return query select 0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint; return;
  end if;
  return query
  with live as (
    select
      count(*) as l_total,
      count(*) filter (where el.status = any(d_set)) as l_delivered,
      count(*) filter (where el.created_at >= now() - interval '24 hours') as l_today,
      count(*) filter (where el.created_at >= now() - interval '48 hours' and el.created_at < now() - interval '24 hours') as l_yesterday,
      count(*) filter (where el.status = any(d_set) and el.created_at >= now() - interval '24 hours') as l_del_today,
      count(*) filter (where el.status = any(d_set) and el.created_at >= now() - interval '48 hours' and el.created_at < now() - interval '24 hours') as l_del_yesterday
    from email_logs el
    where (p_workspace_id is not null and el.workspace_id = p_workspace_id)
       or (p_workspace_id is null and el.user_id = uid)
  ),
  arch as (
    select coalesce(sum(s.total),0) as a_total, coalesce(sum(s.delivered),0) as a_delivered
    from email_stats_daily s
    where p_workspace_id is not null and s.workspace_id = p_workspace_id
  ),
  keys as (
    select count(*) as n from api_keys k
    where k.revoked_at is null
      and ((p_workspace_id is not null and k.workspace_id = p_workspace_id)
        or (p_workspace_id is null and k.user_id = uid))
  )
  select (live.l_total + arch.a_total)::bigint, (live.l_delivered + arch.a_delivered)::bigint, keys.n::bigint,
         live.l_today::bigint, live.l_yesterday::bigint, live.l_del_today::bigint, live.l_del_yesterday::bigint
  from live, arch, keys;
end;
$$;

-- Workspace usage counts in one round-trip (used by /api/settings/usage).
-- emails_month folds in the email_stats_daily retention archive. Gated by
-- workspace membership; identity from auth.uid() (can't be spoofed).
create or replace function public.get_workspace_usage_counts(
  p_workspace_id uuid, p_month_start timestamptz, p_day_start timestamptz
)
returns table(
  emails_month bigint, emails_today bigint, inbound_month bigint,
  domains bigint, api_keys bigint, members bigint, pending_invites bigint,
  templates bigint, webhooks bigint, contacts bigint
)
language plpgsql stable security definer set search_path to 'public'
as $$
declare uid uuid := auth.uid(); allowed boolean;
begin
  allowed := exists (select 1 from workspace_members where workspace_id = p_workspace_id and user_id = uid)
          or exists (select 1 from workspaces where id = p_workspace_id and created_by = uid);
  if not allowed then
    return query select 0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint,0::bigint; return;
  end if;
  return query select
    (coalesce((select count(*) from email_logs where workspace_id = p_workspace_id and created_at >= p_month_start), 0)
     + coalesce((select sum(s.total) from email_stats_daily s where s.workspace_id = p_workspace_id and s.day >= p_month_start::date), 0))::bigint,
    (select count(*) from email_logs where workspace_id = p_workspace_id and created_at >= p_day_start)::bigint,
    (select count(*) from inbound_emails where workspace_id = p_workspace_id and created_at >= p_month_start)::bigint,
    (select count(*) from cloud_domains where workspace_id = p_workspace_id)::bigint,
    (select count(*) from api_keys where workspace_id = p_workspace_id and revoked_at is null and is_mcp = false)::bigint,
    (select count(*) from workspace_members where workspace_id = p_workspace_id)::bigint,
    (select count(*) from workspace_invites where workspace_id = p_workspace_id and status = 'pending')::bigint,
    (select count(*) from email_templates where workspace_id = p_workspace_id)::bigint,
    (select count(*) from webhook_endpoints where workspace_id = p_workspace_id)::bigint,
    (select count(*) from contacts where workspace_id = p_workspace_id)::bigint;
end;
$$;
