BaaS + AI cookbook

-- ai.sources already has RLS enabled by default; tighten the authenticated
-- read policy to only show user's own sources.
DROP POLICY IF EXISTS auth_read_sources ON ai.sources;
CREATE POLICY auth_read_own_sources ON ai.sources
  FOR SELECT TO authenticated
  USING (
    -- Adapt to wherever your "who owns this source" lives — most apps
    -- have a public.documents row with owner_id that joins to ai.sources.id
    EXISTS (
      SELECT 1 FROM public.documents
      WHERE source_id = ai.sources.id
        AND owner_id = auth.uid()
    )
  );

-- ai.chunks inherits the source's ownership.
DROP POLICY IF EXISTS auth_read_chunks ON ai.chunks;
CREATE POLICY auth_read_own_chunks ON ai.chunks
  FOR SELECT TO authenticated
  USING (
    EXISTS (
      SELECT 1 FROM ai.indexed_sources ix
      JOIN public.documents d ON d.source_id = ix.source_id
      WHERE ix.id = ai.chunks.indexed_source_id
        AND d.owner_id = auth.uid()
    )
  );

async function retrieveContext(kbId: string, query: string, userToken: string) {
  // PostgREST does the RLS-filtered hybrid search; we just need top-K chunks
  // ordered by similarity. For real hybrid search, you'd typically POST to
  // /api/knowledge-bases/{kbId}/search instead — but that runs with the
  // service role and skips RLS. Hand-rolling against ai.chunks under the
  // user's JWT gets per-user filtering for free.
  const params = new URLSearchParams({
    select: "id,text,score,source_id,meta",
    knowledge_base_id: `eq.${kbId}`,
    order: "score.desc",
    limit: "10",
  });
  const res = await fetch(`${BASE_URL}/rest/v1/chunks?${params}`, {
    headers: {
      "Accept-Profile": "ai",
      apikey: ANON_KEY,
      Authorization: `Bearer ${userToken}`,
    },
  });
  return res.json();
}

async function askAgent(agentId: string, message: string, contextItems: any[], userToken: string) {
  // The agent run endpoint runs under the service role internally — it does
  // not check RLS on its own tools. Filtering happened above, when we
  // retrieved chunks under the user's JWT.
  const res = await fetch(`${BASE_URL}/api/agents/${agentId}/run`, {
    method: "POST",
    headers: {
      apikey: ANON_KEY,
      Authorization: `Bearer ${SERVICE_ROLE_KEY}`,  // see Recipe 2 about why
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      message,
      context_items: contextItems.map((c) => ({
        text: c.text,
        source_id: c.source_id,
        meta: c.meta,
      })),
    }),
  });
  return res.json();
}

// Backend route. Express-style for illustration.
app.post("/api/chat", requireUserAuth, async (req, res) => {
  const { userId, userJwt } = req.user;  // from your own auth middleware
  const { message } = req.body;

  // Look up the user's agent — or use a shared agent ID for the whole app.
  const agentId = await getAgentForUser(userId);

  // Persist the chat message to your own table for history. RLS-protected
  // public.chat_messages with owner_id = userId enforces per-user history.
  const sessionId = await ensureSession(userId);

  // Stream the agent run from the backend to the client. The platform's SSE
  // stream is preserved by streaming the response body through your handler.
  const agentRes = await fetch(`${POWABASE_URL}/api/agents/${agentId}/run/stream`, {
    method: "POST",
    headers: {
      apikey: SERVICE_ROLE_KEY,
      Authorization: `Bearer ${SERVICE_ROLE_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      message,
      session_id: sessionId,
      // If you need the agent to access user-specific data, pre-fetch and
      // inject as context_items here. See Recipe 1.
    }),
  });

  // Forward the SSE stream to the client.
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  agentRes.body.pipe(res);
});

POST /api/tools
{
  "name": "search_user_articles",
  "description": "Searches the user's saved articles for relevant content.",
  "type": "http",
  "input_schema": {
    "type": "object",
    "properties": {
      "query": { "type": "string" },
      "session_token": { "type": "string" }
    },
    "required": ["query", "session_token"]
  },
  "config": {
    "endpoint": "https://your-app.example.com/internal/tools/search-articles",
    "method": "POST"
  }
}

app.post("/internal/tools/search-articles", async (req, res) => {
  const { query, session_token } = req.body;

  // Look up the user from your opaque session_token — NOT from any header
  // because the platform doesn't forward auth to custom tools.
  const userId = await sessionStore.lookup(session_token);
  if (!userId) return res.status(401).end();

  // Now run the query against your DB with the user's identity applied.
  const results = await db.searchArticles(userId, query);
  res.json({ results });
});

async function uploadAndChat(file: File, userToken: string) {
  // Step 1: Upload to Sources (will be indexed). The user's access token here
  // is for record-keeping; the actual upload runs server-side via your backend
  // if you want auth scoping on what the user can upload.
  const sourceRes = await fetch(`${BASE_URL}/api/sources/upload`, {
    method: "POST",
    headers: {
      apikey: ANON_KEY,
      Authorization: `Bearer ${SERVICE_ROLE_KEY}`,  // Service role — see Recipe 2
    },
    body: (() => {
      const fd = new FormData();
      fd.append("file", file);
      return fd;
    })(),
  });
  const { id: sourceId } = await sourceRes.json();

  // Step 2: Record ownership (so RLS can scope future reads)
  await postgrest("documents", { source_id: sourceId, owner_id: userId });

  // Step 3: Poll for extraction completion (see Sources Reference for shape)
  await pollUntilExtracted(sourceId);

  // Step 4: Add to a user-specific KB
  const kbId = await getUserKb(userId);
  await fetch(`${BASE_URL}/api/knowledge-bases/${kbId}/sources`, {
    method: "POST",
    headers: { apikey: ANON_KEY, Authorization: `Bearer ${SERVICE_ROLE_KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({ source_id: sourceId }),
  });

  // Step 5: Now an agent linked to that KB can answer questions about the PDF.
  return runChat(kbId, userToken);
}

async function streamAndBroadcast(agentId: string, userId: string, message: string) {
  // Open the agent SSE stream
  const agentRes = await fetch(`${POWABASE_URL}/api/agents/${agentId}/run/stream`, { ... });
  const reader = agentRes.body!.getReader();

  // The Realtime topic this user is subscribed to
  const topic = `chat:${userId}`;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const text = decoder.decode(value);
    for (const line of text.split("\n")) {
      if (!line.startsWith("data: ")) continue;
      const event = JSON.parse(line.slice(6));

      // Persist for chat history
      if (event.event === "chunk" || event.event === "complete") {
        await db.chatMessages.insert({ user_id: userId, payload: event });
      }

      // Broadcast for live UIs
      await broadcast({
        topic,
        event: event.event,
        payload: event,
      });
    }
  }
}

CREATE OR REPLACE FUNCTION broadcast_chat_message()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
  PERFORM realtime.send(
    payload  => jsonb_build_object('message_id', NEW.id, 'event', NEW.payload),
    event    => 'chat_event',
    topic    => 'chat:' || NEW.user_id::text,
    private  => true
  );
  RETURN NEW;
END;
$$;

CREATE TRIGGER chat_realtime
  AFTER INSERT ON public.chat_messages
  FOR EACH ROW EXECUTE FUNCTION broadcast_chat_message();