streams example

Author: ianmacartney

convex/_generated/api.d.ts

@@ -21,6 +21,7 @@ import type * as relationshipsExample from "../relationshipsExample.js";
 import type * as retriesExample from "../retriesExample.js";
 import type * as rlsExample from "../rlsExample.js";
 import type * as sessionsExample from "../sessionsExample.js";
+import type * as streamsExample from "../streamsExample.js";
 import type * as testingFunctions from "../testingFunctions.js";
 import type * as triggersExample from "../triggersExample.js";
 
@@ -41,6 +42,7 @@ declare const fullApi: ApiFromModules<{
   retriesExample: typeof retriesExample;
   rlsExample: typeof rlsExample;
   sessionsExample: typeof sessionsExample;
+  streamsExample: typeof streamsExample;
   testingFunctions: typeof testingFunctions;
   triggersExample: typeof triggersExample;
 }>;

convex/schema.ts

@@ -32,4 +32,17 @@ export default defineSchema({
   sum_table: defineTable({ sum: v.number() }),
   notes: defineTable({ session: v.string(), note: v.string() }),
   migrations: migrationsTable,
+  privateMessages: defineTable({
+    from: v.string(),
+    to: v.string(),
+    message: v.string(),
+    // we have creation time, but let's say we want to store it separately
+    sentAt: v.number(),
+  })
+    // inbox
+    .index("to", ["to", "sentAt"])
+    // outbox
+    .index("from", ["from", "sentAt"])
+    // pairwise
+    .index("from_to", ["from", "to", "sentAt"]),
 });

convex/streamsExample.ts

@@ -0,0 +1,86 @@
+import { v } from "convex/values";
+import { api, internal } from "./_generated/api.js";
+import type { Doc, Id } from "./_generated/dataModel.js";
+import {
+  action,
+  internalAction,
+  internalMutation,
+  internalQuery,
+  mutation,
+  query,
+} from "./_generated/server.js";
+import { stream, mergedStream } from "convex-helpers/server/stream";
+import schema from "./schema.js";
+import { paginationOptsValidator } from "convex/server";
+
+export const getInbox = query({
+  args: {
+    id: v.string(),
+    paginationOpts: paginationOptsValidator,
+  },
+  handler: async (ctx, args) => {
+    const messages = await stream(ctx.db, schema)
+      .query("privateMessages")
+      .withIndex("to", (q) => q.eq("to", args.id))
+      .order("desc")
+      .paginate(args.paginationOpts);
+    return messages;
+  },
+});
+
+export const getOutbox = query({
+  args: {
+    id: v.string(),
+    paginationOpts: paginationOptsValidator,
+  },
+  handler: async (ctx, args) => {
+    const messages = await ctx.db
+      .query("privateMessages")
+      .withIndex("from", (q) => q.eq("from", args.id))
+      .order("desc")
+      .paginate(args.paginationOpts);
+    return messages;
+  },
+});
+
+export const getMessagesBetween = query({
+  args: {
+    a: v.string(),
+    b: v.string(),
+    paginationOpts: paginationOptsValidator,
+  },
+  handler: async (ctx, args) => {
+    const aToB = stream(ctx.db, schema)
+      .query("privateMessages")
+      .withIndex("from_to", (q) => q.eq("from", args.a).eq("to", args.b))
+      .order("desc");
+    const bToA = stream(ctx.db, schema)
+      .query("privateMessages")
+      .withIndex("from_to", (q) => q.eq("from", args.b).eq("to", args.a))
+      .order("desc");
+
+    // Both indexes have the "sentAt" field after the fields they're doing
+    // equality on, so they're both sorted by "sentAt" descending, so they
+    // can be merged together.
+    const messages = await mergedStream([aToB, bToA], ["sentAt"]).paginate(
+      args.paginationOpts,
+    );
+    return messages;
+  },
+});
+
+export const sendMessage = mutation({
+  args: {
+    from: v.string(),
+    to: v.string(),
+    message: v.string(),
+  },
+  handler: async (ctx, args) => {
+    await ctx.db.insert("privateMessages", {
+      from: args.from,
+      to: args.to,
+      message: args.message,
+      sentAt: Date.now(),
+    });
+  },
+});

src/App.tsx

@@ -2,6 +2,7 @@ import Counter from "./components/Counter";
 import RelationshipExample from "./components/RelationshipExample";
 import SessionsExample from "./components/SessionsExample";
 import { HonoExample } from "./components/HonoExample";
+import { StreamsExample } from "./components/StreamsExample";
 import { SessionProvider } from "convex-helpers/react/sessions";
 import { CacheExample } from "./components/CacheExample";
 import { ConvexQueryCacheProvider } from "convex-helpers/react/cache";
@@ -21,6 +22,7 @@ export default function App() {
           <SessionsExample />
           <HonoExample />
           <CacheExample />
+          <StreamsExample />
         </ConvexQueryCacheProvider>
       </SessionProvider>
     </main>