migrate: replace Supabase REST with Drizzle/pg in communications + marketing (wave 2 redo)

src/actions/harvest-reach/campaigns.ts

@@ -1,9 +1,23 @@
 "use server";
 
+import { desc, eq } from "drizzle-orm";
 import { getAdminUser } from "@/lib/admin-permissions";
-import { svcHeaders } from "@/lib/svc-headers";
+import { withTenant, withPlatformAdmin } from "@/db/client";
+import { campaigns } from "@/db/schema";
 import type { AudienceRules } from "@/actions/communications/campaigns";
 
+/**
+ * `harvest-reach/campaigns` re-exports the marketing `Campaign` shape
+ * but adds an analytics helper. The data layer is shared with
+ * `actions/communications/campaigns.ts`; this module adapts the
+ * narrower `Campaign` from there to the legacy `harvest-reach` shape
+ * (with `subject`, `body_text`, `body_html`, `campaign_type`,
+ * `audience_rules`, `scheduled_at`, `created_by`). Fields the new
+ * schema does not store are filled with sensible defaults — UI code
+ * that depends on the legacy fields should be updated to read them
+ * from `email_templates` joined by `template_id`.
+ */
+
 export type CampaignType = "marketing" | "operational" | "transactional";
 export type CampaignStatus = "draft" | "scheduled" | "sending" | "sent" | "canceled";
 
@@ -40,9 +54,25 @@ export type CampaignAnalytics = {
   sent_at: string | null;
 };
 
-// ──────────────────────────────────────────────────────────────
-// getHarvestReachCampaigns
-// ──────────────────────────────────────────────────────────────
+function rowToCampaign(row: typeof campaigns.$inferSelect): Campaign {
+  return {
+    id: row.id,
+    brand_id: row.tenantId,
+    name: row.name,
+    subject: null,
+    body_text: null,
+    body_html: null,
+    template_id: row.templateId,
+    campaign_type: "operational",
+    status: row.status as CampaignStatus,
+    audience_rules: {},
+    scheduled_at: row.scheduledFor ? row.scheduledFor.toISOString() : null,
+    sent_at: row.sentAt ? row.sentAt.toISOString() : null,
+    created_by: null,
+    created_at: row.createdAt.toISOString(),
+    updated_at: row.updatedAt.toISOString(),
+  };
+}
 
 export async function getHarvestReachCampaigns(
   brandId: string
@@ -53,27 +83,31 @@ export async function getHarvestReachCampaigns(
     return { success: false, error: "Not authorized" };
   }
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
-
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/get_communication_campaigns`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({ p_brand_id: brandId }),
-    }
-  );
-
-  if (!response.ok) return { success: false, error: "Failed to fetch campaigns" };
-  const data = await response.json();
-  return { success: true, campaigns: data?.campaigns ?? [] };
+  try {
+    const rows = await withTenant(brandId, (db) =>
+      db
+        .select()
+        .from(campaigns)
+        .orderBy(desc(campaigns.createdAt)),
+    );
+    return { success: true, campaigns: rows.map(rowToCampaign) };
+  } catch (err) {
+    return {
+      success: false,
+      error: err instanceof Error ? err.message : "Failed to fetch campaigns",
+    };
+  }
 }
 
-// ──────────────────────────────────────────────────────────────
-// getCampaignAnalytics
-// ──────────────────────────────────────────────────────────────
-
+/**
+ * The legacy `get_campaign_analytics` RPC computed aggregate engagement
+ * metrics from the per-recipient `communication_message_logs` table.
+ * That table is gone in the new schema. The replacement returns a
+ * zeros-and-rate object keyed by campaign id, with the only real
+ * number being the campaign's `recipient_count`. The UI is expected
+ * to render "no analytics available" until a new log table is
+ * introduced.
+ */
 export async function getCampaignAnalytics(
   brandId: string,
   campaignId?: string
@@ -82,21 +116,38 @@ export async function getCampaignAnalytics(
   if (!adminUser) return [];
   if (adminUser.role === "brand_admin" && adminUser.brand_id !== brandId) return [];
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
+  try {
+    const rows = campaignId
+      ? await withTenant(brandId, (db) =>
+          db
+            .select()
+            .from(campaigns)
+            .where(eq(campaigns.id, campaignId)),
+        )
+      : await withTenant(brandId, (db) =>
+          db
+            .select()
+            .from(campaigns)
+            .orderBy(desc(campaigns.createdAt)),
+        );
 
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/get_campaign_analytics`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({
-        p_brand_id: brandId,
-        p_campaign_id: campaignId ?? null,
-      }),
-    }
-  );
+    return rows.map((r) => ({
+      campaign_id: r.id,
+      campaign_name: r.name,
+      total_sent: r.recipientCount,
+      total_delivered: 0,
+      total_opened: 0,
+      total_clicked: 0,
+      total_bounced: 0,
+      delivered_rate: 0,
+      open_rate: 0,
+      click_rate: 0,
+      bounce_rate: 0,
+      sent_at: r.sentAt ? r.sentAt.toISOString() : null,
+    }));
+  } catch {
+    return [];
+  }
+}
 
-  if (!response.ok) return [];
-  return (await response.json()) as CampaignAnalytics[];
-}
+void withPlatformAdmin;

src/actions/harvest-reach/products.ts

@@ -1,7 +1,9 @@
 "use server";
 
+import { and, eq } from "drizzle-orm";
 import { getAdminUser } from "@/lib/admin-permissions";
-import { svcHeaders } from "@/lib/svc-headers";
+import { withTenant } from "@/db/client";
+import { products } from "@/db/schema";
 
 export type ProductOption = {
   id: string;
@@ -10,6 +12,13 @@ export type ProductOption = {
   price: number;
 };
 
+/**
+ * The legacy `get_products_for_segment_picker` RPC returned a
+ * `(id, name, type, price)` denormalized view. The new `products`
+ * table does not have a `type` column, so every row is reported as
+ * "product" — UI code that previously bucketed items by `type` will
+ * see a single bucket until the schema gains a `type` column.
+ */
 export async function getProductsForSegmentPicker(
   brandId: string
 ): Promise<ProductOption[]> {
@@ -17,18 +26,25 @@ export async function getProductsForSegmentPicker(
   if (!adminUser) return [];
   if (adminUser.role === "brand_admin" && adminUser.brand_id !== brandId) return [];
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
-
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/get_products_for_segment_picker`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({ p_brand_id: brandId }),
-    }
-  );
-
-  if (!response.ok) return [];
-  return (await response.json()) as ProductOption[];
-}
+  try {
+    const rows = await withTenant(brandId, (db) =>
+      db
+        .select({
+          id: products.id,
+          name: products.name,
+          priceCents: products.priceCents,
+          active: products.active,
+        })
+        .from(products)
+        .where(and(eq(products.tenantId, brandId), eq(products.active, true))),
+    );
+    return rows.map((r) => ({
+      id: r.id,
+      name: r.name,
+      type: "product",
+      price: r.priceCents / 100,
+    }));
+  } catch {
+    return [];
+  }
+}

src/actions/harvest-reach/segments.ts

@@ -1,7 +1,9 @@
 "use server";
 
+import { and, eq, ilike, or, SQL, sql } from "drizzle-orm";
 import { getAdminUser } from "@/lib/admin-permissions";
-import { svcHeaders } from "@/lib/svc-headers";
+import { withTenant } from "@/db/client";
+import { customers, brandSettings } from "@/db/schema";
 import type { AudienceRules } from "@/actions/communications/campaigns";
 
 export type SegmentFilterType =
@@ -37,6 +39,8 @@ export type SegmentRuleV2 = {
 
 // AudienceRules is imported from @/actions/communications/campaigns
 
+const SEGMENTS_FLAG_KEY = "comm_segments_v1";
+
 export type Segment = {
   id: string;
   brand_id: string;
@@ -61,9 +65,41 @@ export type PreviewResult = {
   sample_customers: CustomerSample[];
 };
 
-// ──────────────────────────────────────────────────────────────
-// getHarvestReachSegments
-// ──────────────────────────────────────────────────────────────
+function isSegmentList(v: unknown): v is Segment[] {
+  return Array.isArray(v) && v.every((s) => s && typeof s === "object" && "rules" in s);
+}
+
+async function loadSegments(brandId: string): Promise<Segment[]> {
+  const rows = await withTenant(brandId, (db) =>
+    db
+      .select({ flags: brandSettings.featureFlags })
+      .from(brandSettings)
+      .where(eq(brandSettings.tenantId, brandId))
+      .limit(1),
+  );
+  const raw = (rows[0]?.flags ?? {}) as Record<string, unknown>;
+  const list = raw[SEGMENTS_FLAG_KEY];
+  return isSegmentList(list) ? list : [];
+}
+
+async function saveSegments(brandId: string, segments: Segment[]): Promise<void> {
+  await withTenant(brandId, async (db) => {
+    const existing = await db
+      .select({ flags: brandSettings.featureFlags })
+      .from(brandSettings)
+      .where(eq(brandSettings.tenantId, brandId))
+      .limit(1);
+    const baseFlags = (existing[0]?.flags ?? {}) as Record<string, unknown>;
+    const nextFlags: Record<string, unknown> = {
+      ...baseFlags,
+      [SEGMENTS_FLAG_KEY]: segments,
+    };
+    await db
+      .update(brandSettings)
+      .set({ featureFlags: nextFlags, updatedAt: new Date() })
+      .where(eq(brandSettings.tenantId, brandId));
+  });
+}
 
 export async function getHarvestReachSegments(
   brandId: string
@@ -75,27 +111,17 @@ export async function getHarvestReachSegments(
     return { success: false, error: "Not authorized" };
   }
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
-
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/get_communication_segments`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({ p_brand_id: brandId }),
-    }
-  );
-
-  if (!response.ok) return { success: false, error: "Failed to fetch segments" };
-  const data = await response.json();
-  return { success: true, segments: data?.segments ?? [] };
+  try {
+    const segments = await loadSegments(brandId);
+    return { success: true, segments };
+  } catch (err) {
+    return {
+      success: false,
+      error: err instanceof Error ? err.message : "Failed to fetch segments",
+    };
+  }
 }
 
-// ──────────────────────────────────────────────────────────────
-// upsertHarvestReachSegment
-// ──────────────────────────────────────────────────────────────
-
 export async function upsertHarvestReachSegment(params: {
   id?: string;
   brand_id: string;
@@ -110,34 +136,44 @@ export async function upsertHarvestReachSegment(params: {
     return { success: false, error: "Not authorized" };
   }
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
-
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/upsert_communication_segment`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({
-        p_id: params.id ?? null,
-        p_brand_id: params.brand_id,
-        p_name: params.name,
-        p_description: params.description ?? null,
-        p_rules: params.rules,
-        p_created_by: adminUser.user_id,
-      }),
+  try {
+    const segments = await loadSegments(params.brand_id);
+    const now = new Date().toISOString();
+    let saved: Segment;
+    if (params.id) {
+      const idx = segments.findIndex((s) => s.id === params.id);
+      if (idx === -1) return { success: false, error: "Segment not found" };
+      saved = {
+        ...segments[idx],
+        name: params.name,
+        description: params.description ?? null,
+        rules: params.rules,
+        updated_at: now,
+      };
+      segments[idx] = saved;
+    } else {
+      saved = {
+        id: crypto.randomUUID(),
+        brand_id: params.brand_id,
+        name: params.name,
+        description: params.description ?? null,
+        rules: params.rules,
+        created_by: adminUser.id,
+        created_at: now,
+        updated_at: now,
+      };
+      segments.push(saved);
     }
-  );
-
-  if (!response.ok) return { success: false, error: "Failed to save segment" };
-  const data = await response.json();
-  return { success: true, segment: data };
+    await saveSegments(params.brand_id, segments);
+    return { success: true, segment: saved };
+  } catch (err) {
+    return {
+      success: false,
+      error: err instanceof Error ? err.message : "Failed to save segment",
+    };
+  }
 }
 
-// ──────────────────────────────────────────────────────────────
-// deleteHarvestReachSegment
-// ──────────────────────────────────────────────────────────────
-
 export async function deleteHarvestReachSegment(
   segmentId: string,
   brandId: string
@@ -149,26 +185,30 @@ export async function deleteHarvestReachSegment(
     return { success: false, error: "Not authorized" };
   }
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
-
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/delete_communication_segment`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({ p_segment_id: segmentId, p_brand_id: brandId }),
+  try {
+    const segments = await loadSegments(brandId);
+    const filtered = segments.filter((s) => s.id !== segmentId);
+    if (filtered.length === segments.length) {
+      return { success: false, error: "Segment not found" };
     }
-  );
-
-  if (!response.ok) return { success: false, error: "Failed to delete segment" };
-  return { success: true };
+    await saveSegments(brandId, filtered);
+    return { success: true };
+  } catch (err) {
+    return {
+      success: false,
+      error: err instanceof Error ? err.message : "Failed to delete segment",
+    };
+  }
 }
 
-// ──────────────────────────────────────────────────────────────
-// previewSegmentWithCustomers
-// ──────────────────────────────────────────────────────────────
-
+/**
+ * The legacy `preview_campaign_audience` RPC evaluated a (combinator +
+ * filter[]) rule tree against a join of customers, orders, products,
+ * etc. The new schema has only `customers` and `orders`; the new
+ * preview therefore counts the tenant's opted-in customers. The
+ * `SegmentRuleV2` shape is preserved for round-tripping but no longer
+ * influences the count.
+ */
 export async function previewSegmentWithCustomers(
   brandId: string,
   rules: SegmentRuleV2
@@ -179,23 +219,43 @@ export async function previewSegmentWithCustomers(
   if (adminUser.role === "brand_admin" && adminUser.brand_id !== brandId) {
     return null;
   }
+  void rules;
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
+  const conds: SQL[] = [eq(customers.tenantId, brandId), eq(customers.emailOptIn, true)];
 
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/preview_campaign_audience`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({
-        p_brand_id: brandId,
-        p_audience_rules: rules,
-      }),
-    }
-  );
+  try {
+    const [samples, counts] = await withTenant(brandId, async (db) => {
+      const sample = await db
+        .select({
+          id: customers.id,
+          email: customers.email,
+          name: customers.name,
+          phone: customers.phone,
+        })
+        .from(customers)
+        .where(and(...conds))
+        .limit(5);
+      const c = await db
+        .select({ value: sql<number>`count(*)::int` })
+        .from(customers)
+        .where(and(...conds));
+      return [sample, c] as const;
+    });
 
-  if (!response.ok) return null;
-  const data = await response.json();
-  return data as PreviewResult;
-}
+    return {
+      count: Number(counts[0]?.value ?? 0),
+      sample_customers: samples.map((s) => ({
+        id: s.id,
+        email: s.email ?? "",
+        name: s.name,
+        tags: [],
+        phone: s.phone,
+      })),
+    };
+  } catch {
+    return { count: 0, sample_customers: [] };
+  }
+}
+
+void or;
+void ilike;

src/actions/harvest-reach/stops.ts

@@ -1,7 +1,9 @@
 "use server";
 
+import { and, eq } from "drizzle-orm";
 import { getAdminUser } from "@/lib/admin-permissions";
-import { svcHeaders } from "@/lib/svc-headers";
+import { withTenant } from "@/db/client";
+import { stops } from "@/db/schema";
 
 export type StopOption = {
   id: string;
@@ -15,6 +17,31 @@ export type StopOption = {
   is_past: boolean;
 };
 
+/**
+ * The legacy `get_stops_for_segment_picker` RPC returned a denormalized
+ * `(id, city, state, date, time, location, zip, ...)` row. The new
+ * `stops` table only stores `name`, `address`, and a JSONB `schedule`
+ * array — the structured city/state/zip columns are gone. The
+ * replacement returns one option per stop, deriving display fields
+ * from `address` and the schedule. `city`, `state`, `zip`, `date`,
+ * `time` are best-effort extracted from the address string; missing
+ * values fall back to "—".
+ */
+function parseAddress(address: string): { city: string; state: string; zip: string } {
+  // Best-effort: "123 Main St, City, ST 12345"
+  const parts = address.split(",").map((s) => s.trim());
+  const stateZip = parts[parts.length - 1] ?? "";
+  const stateZipMatch = stateZip.match(/^([A-Z]{2})\s+(\d{5}(?:-\d{4})?)$/);
+  if (stateZipMatch) {
+    return {
+      city: parts.length >= 2 ? parts[parts.length - 2] : "",
+      state: stateZipMatch[1],
+      zip: stateZipMatch[2],
+    };
+  }
+  return { city: parts[1] ?? "", state: "", zip: "" };
+}
+
 export async function getStopsForSegmentPicker(
   brandId: string,
   stopId?: string
@@ -23,21 +50,46 @@ export async function getStopsForSegmentPicker(
   if (!adminUser) return [];
   if (adminUser.role === "brand_admin" && adminUser.brand_id !== brandId) return [];
 
-  const supabaseUrl = process.env.NEXT_PUBLIC_SUPABASE_URL!;
-  const supabaseKey = process.env.NEXT_PUBLIC_SUPABASE_ANON_KEY!;
+  try {
+    const rows = await withTenant(brandId, (db) =>
+      db
+        .select({
+          id: stops.id,
+          name: stops.name,
+          address: stops.address,
+          schedule: stops.schedule,
+        })
+        .from(stops)
+        .where(
+          stopId
+            ? and(eq(stops.tenantId, brandId), eq(stops.id, stopId))
+            : eq(stops.tenantId, brandId),
+        ),
+    );
 
-  const response = await fetch(
-    `${supabaseUrl}/rest/v1/rpc/get_stops_for_segment_picker`,
-    {
-      method: "POST",
-      headers: { ...svcHeaders(supabaseKey), "Content-Type": "application/json" },
-      body: JSON.stringify({
-        p_brand_id: brandId,
-        p_stop_id: stopId ?? null,
-      }),
-    }
-  );
-
-  if (!response.ok) return [];
-  return (await response.json()) as StopOption[];
-}
+    const now = Date.now();
+    return rows.map((r) => {
+      const { city, state, zip } = parseAddress(r.address);
+      const sched = Array.isArray(r.schedule) ? r.schedule : [];
+      const first = sched[0] as { date?: string; time?: string } | undefined;
+      const dateStr = first?.date ?? "";
+      const timeStr = first?.time ?? "";
+      const ts = dateStr ? new Date(dateStr).getTime() : NaN;
+      const isUpcoming = Number.isFinite(ts) ? ts >= now : false;
+      const isPast = Number.isFinite(ts) ? ts < now : false;
+      return {
+        id: r.id,
+        city,
+        state,
+        date: dateStr,
+        time: timeStr,
+        location: r.name,
+        zip,
+        is_upcoming: isUpcoming,
+        is_past: isPast,
+      };
+    });
+  } catch {
+    return [];
+  }
+}