feat(flows): community flows — publish / browse / import (core)

apps/api/drizzle/migrations/0071_community_flows_hub.sql

@@ -0,0 +1,54 @@
+-- 0071 — Community Flows hub. A CENTRAL, cross-instance catalog of shareable
+-- flow templates (n8n-style). Populated only on the SaaS hub (EE); inert empty
+-- tables on self-host. Deliberately NOT workspace-scoped and NO RLS — the hub
+-- serves cross-instance reads via the superuser db client (mirrors docs.public).
+-- Idempotent (drizzle ledger is behind — apply via psql).
+
+CREATE TABLE IF NOT EXISTS "community_flows" (
+  "id"               uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  "slug"             text NOT NULL,
+  "name"             text NOT NULL,
+  "description"      text,
+  "tags"             text[] NOT NULL DEFAULT ARRAY[]::text[],
+  "template_json"    jsonb NOT NULL,
+  "schema_version"   integer NOT NULL DEFAULT 1,
+  "publisher_email"  text NOT NULL,
+  "publisher_handle" text,
+  "source_version"   text,
+  "node_count"       integer NOT NULL DEFAULT 0,
+  "edge_count"       integer NOT NULL DEFAULT 0,
+  "import_count"     integer NOT NULL DEFAULT 0,
+  "status"           text NOT NULL DEFAULT 'listed',
+  "created_at"       timestamptz NOT NULL DEFAULT now(),
+  "updated_at"       timestamptz NOT NULL DEFAULT now()
+);
+
+DO $$ BEGIN
+  ALTER TABLE "community_flows" ADD CONSTRAINT "community_flows_slug_key" UNIQUE ("slug");
+EXCEPTION WHEN duplicate_table THEN NULL; WHEN duplicate_object THEN NULL; END $$;
+
+DO $$ BEGIN
+  ALTER TABLE "community_flows"
+    ADD CONSTRAINT "community_flows_publisher_name_key" UNIQUE ("publisher_email", "name");
+EXCEPTION WHEN duplicate_table THEN NULL; WHEN duplicate_object THEN NULL; END $$;
+
+CREATE INDEX IF NOT EXISTS "community_flows_status_idx"     ON "community_flows" ("status");
+CREATE INDEX IF NOT EXISTS "community_flows_publisher_idx"  ON "community_flows" ("publisher_email");
+CREATE INDEX IF NOT EXISTS "community_flows_popularity_idx" ON "community_flows" ("import_count");
+CREATE INDEX IF NOT EXISTS "community_flows_tags_gin"       ON "community_flows" USING gin ("tags");
+
+CREATE TABLE IF NOT EXISTS "community_flow_reports" (
+  "id"                   uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  "community_flow_id"    uuid NOT NULL REFERENCES "community_flows"("id") ON DELETE CASCADE,
+  "reason"               text NOT NULL,
+  "detail"               text,
+  "reporter_fingerprint" text,
+  "created_at"           timestamptz NOT NULL DEFAULT now(),
+  "resolved_at"          timestamptz,
+  "resolved_by"          text
+);
+
+CREATE INDEX IF NOT EXISTS "community_flow_reports_flow_idx"
+  ON "community_flow_reports" ("community_flow_id");
+CREATE INDEX IF NOT EXISTS "community_flow_reports_unresolved_idx"
+  ON "community_flow_reports" ("resolved_at");

apps/api/drizzle/migrations/0072_flow_community_slug.sql

@@ -0,0 +1,7 @@
+-- 0072 — Community Flows: track whether a local flow is published to the hub.
+-- When a workspace publishes a flow to the community, we store the returned hub
+-- slug here so the UI can show "published to community" state + an unpublish
+-- affordance, and badge the flow on the list page. Core migration (flows is a
+-- core table) — applies on self-host too. Idempotent (apply via psql).
+
+ALTER TABLE "flows" ADD COLUMN IF NOT EXISTS "community_slug" text;

apps/api/src/config/env.ts

@@ -244,6 +244,17 @@ const envSchema = z.object({
   // Ed25519 private signing key (PEM) — ISSUING SIDE ONLY (cloud). Absent on
   // self-host, which only verifies against the baked-in public key.
   LICENSE_SIGNING_KEY: z.string().optional(),
+
+  // Community Flows hub (open-core). Every instance — cloud AND self-hosted —
+  // talks to a CENTRAL hub to browse/import/publish flow templates. Defaults to
+  // the SaaS host; self-hosters may point elsewhere or disable entirely.
+  //   URL      — base URL of the hub API (serves /api/hub/*).
+  //   ENABLED  — false fully disables the feature (air-gapped self-host).
+  //   KEY      — the instance's community-license key, used only to authenticate
+  //              PUBLISH calls. Browse/import need no key. Absent = publish off.
+  COMMUNITY_HUB_URL:     z.string().url().default('https://api.theboringpeople.in'),
+  COMMUNITY_HUB_ENABLED: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'),
+  COMMUNITY_HUB_KEY:     z.string().min(1).optional(),
 });
 
 const parsed = envSchema

apps/api/src/db/schema.ts

@@ -961,6 +961,9 @@ export const flows = pgTable(
     // the link can view the published flow read-only.
     shareToken: text('share_token'),
     sharedAt:   timestamp('shared_at', { withTimezone: true }),
+    // Community Flows (0072) — set to the hub listing slug when this flow has
+    // been published to the community hub; null when not published.
+    communitySlug: text('community_slug'),
   },
   (table) => ({
     workspaceSlugUnique: unique('flows_workspace_id_slug_key').on(table.workspaceId, table.slug),
@@ -1801,6 +1804,68 @@ export const communityRegistrations = pgTable('community_registrations', {
   createdAt:       timestamp('created_at', { withTimezone: true }).notNull().default(sql`now()`),
 });
 
+/* ── Community Flows hub (migration 0071) ─────────────────────────────────────
+ * CENTRAL, cross-instance catalog of shareable flow templates. Unlike almost
+ * everything else, these tables are NOT workspace-scoped and carry NO RLS — the
+ * hub is cross-instance by design and is read via the superuser `db` client
+ * (mirrors the /api/docs/public pattern). Populated only on the SaaS hub (EE);
+ * inert empty tables on self-host installs. template_json holds the portable
+ * PortableFlowTemplate (lib/flows/portability.ts) with all workspace ids stripped. */
+export const communityFlows = pgTable(
+  'community_flows',
+  {
+    id: uuid('id').primaryKey().default(sql`gen_random_uuid()`),
+    slug: text('slug').notNull().unique(),
+    name: text('name').notNull(),
+    description: text('description'),
+    tags: text('tags').array().notNull().default(sql`ARRAY[]::text[]`),
+    templateJson: jsonb('template_json').notNull(),
+    schemaVersion: integer('schema_version').notNull().default(1),
+    // Publisher identity derived from the validated community-license key.
+    publisherEmail: text('publisher_email').notNull(),
+    publisherHandle: text('publisher_handle'),
+    sourceVersion: text('source_version'),
+    nodeCount: integer('node_count').notNull().default(0),
+    edgeCount: integer('edge_count').notNull().default(0),
+    importCount: integer('import_count').notNull().default(0),
+    // listed | unlisted | removed
+    status: text('status').notNull().default('listed'),
+    createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
+    updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(),
+  },
+  (table) => ({
+    statusIdx: index('community_flows_status_idx').on(table.status),
+    publisherIdx: index('community_flows_publisher_idx').on(table.publisherEmail),
+    popularityIdx: index('community_flows_popularity_idx').on(table.importCount),
+    // one listing per (publisher, name) — re-publishing updates in place.
+    publisherNameUnique: unique('community_flows_publisher_name_key').on(
+      table.publisherEmail,
+      table.name,
+    ),
+  }),
+);
+
+export const communityFlowReports = pgTable(
+  'community_flow_reports',
+  {
+    id: uuid('id').primaryKey().default(sql`gen_random_uuid()`),
+    communityFlowId: uuid('community_flow_id')
+      .notNull()
+      .references(() => communityFlows.id, { onDelete: 'cascade' }),
+    // spam | malicious | inappropriate | broken | other
+    reason: text('reason').notNull(),
+    detail: text('detail'),
+    reporterFingerprint: text('reporter_fingerprint'),
+    createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
+    resolvedAt: timestamp('resolved_at', { withTimezone: true }),
+    resolvedBy: text('resolved_by'),
+  },
+  (table) => ({
+    flowIdx: index('community_flow_reports_flow_idx').on(table.communityFlowId),
+    unresolvedIdx: index('community_flow_reports_unresolved_idx').on(table.resolvedAt),
+  }),
+);
+
 // ── Internal admin center (migration 0047) ──────────────────────────────────────
 
 /** A license: a plan tier + seats + entitlements, optionally bound to a workspace

apps/api/src/lib/community/hub-client.ts

@@ -0,0 +1,158 @@
+/**
+ * Community hub client (Community Flows — Phase 2, P2-1).
+ *
+ * The CORE-side HTTP client every instance uses to talk to the central hub
+ * (COMMUNITY_HUB_URL). This is what makes a self-hosted install participate in
+ * the same catalog as the cloud. Reads retry on transient errors; writes never
+ * retry (publish must not double-post). When the feature is disabled, every
+ * call throws CommunityDisabledError so routes can translate to 404.
+ */
+import { getHubBaseUrl, getHubKey, isCommunityEnabled } from './hub-config.js';
+import { config } from '../../config/env.js';
+
+export class CommunityDisabledError extends Error {
+  constructor() {
+    super('The community hub is disabled on this instance.');
+    this.name = 'CommunityDisabledError';
+  }
+}
+
+export class HubError extends Error {
+  constructor(
+    public status: number,
+    public code: string,
+    message: string,
+    public body?: unknown,
+  ) {
+    super(message);
+    this.name = 'HubError';
+  }
+}
+
+const TIMEOUT_MS = 10_000;
+
+interface FetchOpts {
+  method?: string;
+  body?: unknown;
+  key?: string;
+  retries?: number;
+}
+
+async function hubFetch<T>(path: string, opts: FetchOpts = {}): Promise<T> {
+  if (!isCommunityEnabled()) throw new CommunityDisabledError();
+  const url = `${getHubBaseUrl()}${path}`;
+  const method = opts.method ?? 'GET';
+  const retries = opts.retries ?? (method === 'GET' ? 2 : 0);
+
+  let lastErr: unknown;
+  for (let attempt = 0; attempt <= retries; attempt++) {
+    const ac = new AbortController();
+    const timer = setTimeout(() => ac.abort(), TIMEOUT_MS);
+    try {
+      const res = await fetch(url, {
+        method,
+        headers: {
+          'content-type': 'application/json',
+          ...(opts.key ? { authorization: `Bearer ${opts.key}` } : {}),
+          'x-mnema-instance': config.WEB_BASE_URL,
+        },
+        body: opts.body !== undefined ? JSON.stringify(opts.body) : undefined,
+        signal: ac.signal,
+      });
+      clearTimeout(timer);
+
+      const text = await res.text();
+      const parsed = text ? (JSON.parse(text) as unknown) : null;
+      if (!res.ok) {
+        // Retry 5xx on reads only; surface 4xx immediately.
+        if (res.status >= 500 && attempt < retries) {
+          lastErr = new HubError(res.status, 'hub_5xx', `Hub returned ${res.status}`, parsed);
+          continue;
+        }
+        const code = (parsed as { error?: string })?.error ?? `http_${res.status}`;
+        throw new HubError(res.status, code, `Hub error ${res.status}`, parsed);
+      }
+      return parsed as T;
+    } catch (err) {
+      clearTimeout(timer);
+      if (err instanceof HubError) throw err;
+      // Network/abort — retry on reads.
+      lastErr = err;
+      if (attempt >= retries) throw new HubError(502, 'hub_unreachable', 'Community hub is unreachable.', String(err));
+    }
+  }
+  throw lastErr instanceof Error ? lastErr : new HubError(502, 'hub_unreachable', 'Community hub is unreachable.');
+}
+
+export interface HubListItem {
+  slug: string;
+  name: string;
+  description: string | null;
+  tags: string[];
+  nodeCount: number;
+  importCount: number;
+  publisherHandle: string | null;
+  createdAt: string;
+}
+
+export interface HubListResult {
+  flows: HubListItem[];
+  next_cursor: number | null;
+}
+
+export interface HubFlowDetail {
+  slug: string;
+  name: string;
+  description: string | null;
+  tags: string[];
+  schema_version: number;
+  template_json: unknown;
+  node_count: number;
+  edge_count: number;
+  import_count: number;
+  publisher_handle: string | null;
+  created_at: string;
+}
+
+export function hubList(params: {
+  q?: string;
+  tag?: string;
+  sort?: string;
+  cursor?: number;
+  limit?: number;
+}): Promise<HubListResult> {
+  const qs = new URLSearchParams();
+  if (params.q) qs.set('q', params.q);
+  if (params.tag) qs.set('tag', params.tag);
+  if (params.sort) qs.set('sort', params.sort);
+  if (params.cursor) qs.set('cursor', String(params.cursor));
+  if (params.limit) qs.set('limit', String(params.limit));
+  const suffix = qs.toString();
+  return hubFetch<HubListResult>(`/api/hub/flows${suffix ? `?${suffix}` : ''}`);
+}
+
+export function hubGet(slug: string): Promise<HubFlowDetail> {
+  return hubFetch<HubFlowDetail>(`/api/hub/flows/${encodeURIComponent(slug)}`);
+}
+
+export function hubPublish(
+  body: { template_json: unknown; name: string; description: string | null; tags: string[]; source_version?: string },
+): Promise<{ slug: string; url: string }> {
+  const key = getHubKey();
+  if (!key) throw new HubError(400, 'no_key', 'No community key configured on this instance.');
+  return hubFetch(`/api/hub/flows`, { method: 'POST', body, key });
+}
+
+export function hubUnpublish(slug: string): Promise<{ removed: boolean }> {
+  const key = getHubKey();
+  if (!key) throw new HubError(400, 'no_key', 'No community key configured on this instance.');
+  return hubFetch(`/api/hub/flows/${encodeURIComponent(slug)}`, { method: 'DELETE', key });
+}
+
+export function hubReport(slug: string, body: { reason: string; detail?: string }): Promise<{ received: boolean }> {
+  return hubFetch(`/api/hub/flows/${encodeURIComponent(slug)}/report`, { method: 'POST', body });
+}
+
+export function hubBumpImport(slug: string): Promise<{ ok: boolean }> {
+  return hubFetch(`/api/hub/flows/${encodeURIComponent(slug)}/imported`, { method: 'POST', retries: 0 });
+}

apps/api/src/lib/community/hub-config.ts

@@ -0,0 +1,29 @@
+/**
+ * Community hub configuration helpers (Community Flows — Phase 0, P0-4).
+ *
+ * A thin typed surface over the COMMUNITY_HUB_* env so the rest of the code
+ * never reads process.env directly. Shared by the hub client (P2-1), the proxy
+ * routes (P2-2), and the publish route (P2-4).
+ */
+import { config } from '../../config/env.js';
+
+/** Whether the community feature is enabled on this instance at all. */
+export function isCommunityEnabled(): boolean {
+  return config.COMMUNITY_HUB_ENABLED;
+}
+
+/** Base URL of the central hub (no trailing slash). */
+export function getHubBaseUrl(): string {
+  return config.COMMUNITY_HUB_URL.replace(/\/+$/, '');
+}
+
+/** This instance's community-license key, used to authenticate publishes.
+ *  Undefined when unset — publish is disabled with a clear message. */
+export function getHubKey(): string | undefined {
+  return config.COMMUNITY_HUB_KEY;
+}
+
+/** True when this instance can publish (feature on + a key configured). */
+export function canPublishToCommunity(): boolean {
+  return isCommunityEnabled() && !!getHubKey();
+}