convex-helpers-patterns

Convex Helpers ライブラリのパターン

概要

convex-helpers ライブラリは、一般的な Convex 開発のニーズに対応する、実戦で検証済みのパターンを提供します。このスキルでは、トリガー（自動的な副作用）、行レベルセキュリティ、リレーションシップヘルパー、カスタム関数、レート制限、並行処理を制御する Workpool を扱います。

インストール

npm install convex-helpers @convex-dev/workpool

TypeScript：any 型は絶対に使用しない

重要なルール： このコードベースでは @typescript-eslint/no-explicit-any が有効になっています。any を使用するとビルドに失敗します。

このスキルを使用する場面

以下の場合にこのスキルを使用します。

• ドキュメントの変更時に自動的な副作用を実装する（トリガー）
• 宣言的なアクセス制御を追加する（行レベルセキュリティ）
• ドキュメント間のリレーションシップをたどる
• 再利用可能な認証済み関数ラッパーを作成する
• レート制限を実装する
• Workpool で同時書き込みを管理する
• カスタム関数ビルダーを作成する

主なパターンの概要

パターン	ユースケース
トリガー	ドキュメントの変更時にコードを自動実行する
行レベルセキュリティ	データベース層で宣言的なアクセス制御を行う
リレーションシップヘルパー	ドキュメント間のリレーションを簡単にたどる
カスタム関数	クエリ／ミューテーションを認証やログ記録などでラップする
レートリミッター	アプリケーションレベルのレート制限を行う
Workpool	並列ジョブを分散実行し、競合するミューテーションを直列化する
マイグレーション	状態を追跡しながらスキーマを移行する

トリガー（自動的な副作用）

トリガーは、ドキュメントの変更時にコードを自動実行します。ミューテーションと同じトランザクション内でアトミックに実行されます。

トリガーの設定

// convex/functions.ts
import { mutation as rawMutation } from "./_generated/server";
import { Triggers } from "convex-helpers/server/triggers";
import {
  customCtx,
  customMutation,
} from "convex-helpers/server/customFunctions";
import { DataModel } from "./_generated/dataModel";

const triggers = new Triggers<DataModel>();

// 1. Compute fullName on every user change
triggers.register("users", async (ctx, change) => {
  if (change.newDoc) {
    const fullName = `${change.newDoc.firstName} ${change.newDoc.lastName}`;
    if (change.newDoc.fullName !== fullName) {
      await ctx.db.patch(change.id, { fullName });
    }
  }
});

// 2. Keep denormalized count (careful: single doc = write contention)
triggers.register("users", async (ctx, change) => {
  const countDoc = (await ctx.db.query("userCount").unique())!;
  if (change.operation === "insert") {
    await ctx.db.patch(countDoc._id, { count: countDoc.count + 1 });
  } else if (change.operation === "delete") {
    await ctx.db.patch(countDoc._id, { count: countDoc.count - 1 });
  }
});

// 3. Cascading deletes
triggers.register("users", async (ctx, change) => {
  if (change.operation === "delete") {
    const messages = await ctx.db
      .query("messages")
      .withIndex("by_author", (q) => q.eq("authorId", change.id))
      .collect();
    for (const msg of messages) {
      await ctx.db.delete(msg._id);
    }
  }
});

// Export wrapped mutation that runs triggers
export const mutation = customMutation(rawMutation, customCtx(triggers.wrapDB));

トリガーの変更オブジェクト

interface Change<Doc> {
  id: Id<TableName>;
  operation: "insert" | "update" | "delete";
  oldDoc: Doc | null; // null for inserts
  newDoc: Doc | null; // null for deletes
}

トリガーに関する警告

警告： トリガーはミューテーションと同じトランザクション内で実行されます。トリガー内でホットスポットとなるドキュメント（グローバルカウンターなど）に書き込むと、高負荷時に OCC の競合が発生します。競合が多い書き込みには、シャーディングまたは Workpool を使用してください。

行レベルセキュリティ（RLS）

データベース層で宣言的なアクセス制御を行います。RLS はデータベースコンテキストをラップし、すべての読み取りと書き込みにルールを適用します。

RLS の設定

// convex/functions.ts
import {
  Rules,
  wrapDatabaseReader,
  wrapDatabaseWriter,
} from "convex-helpers/server/rowLevelSecurity";
import {
  customCtx,
  customQuery,
  customMutation,
} from "convex-helpers/server/customFunctions";
import { query, mutation } from "./_generated/server";
import { QueryCtx } from "./_generated/server";
import { DataModel } from "./_generated/dataModel";

async function rlsRules(ctx: QueryCtx) {
  const identity = await ctx.auth.getUserIdentity();

  return {
    users: {
      read: async (_, user) => {
        // Unauthenticated users can only read users over 18
        if (!identity && user.age < 18) return false;
        return true;
      },
      insert: async () => true,
      modify: async (_, user) => {
        if (!identity) throw new Error("Must be authenticated");
        // Users can only modify their own record
        return user.tokenIdentifier === identity.tokenIdentifier;
      },
    },

    messages: {
      read: async (_, message) => {
        // Only read messages in conversations you're a member of
        const conversation = await ctx.db.get(message.conversationId);
        return conversation?.members.includes(identity?.subject ?? "") ?? false;
      },
      modify: async (_, message) => {
        // Only modify your own messages
        return message.authorId === identity?.subject;
      },
    },

    // Table with no restrictions
    publicPosts: {
      read: async () => true,
      insert: async () => true,
      modify: async () => true,
    },
  } satisfies Rules<QueryCtx, DataModel>;
}

// Wrap query/mutation with RLS
export const queryWithRLS = customQuery(
  query,
  customCtx(async (ctx) => ({
    db: wrapDatabaseReader(ctx, ctx.db, await rlsRules(ctx)),
  }))
);

export const mutationWithRLS = customMutation(
  mutation,
  customCtx(async (ctx) => ({
    db: wrapDatabaseWriter(ctx, ctx.db, await rlsRules(ctx)),
  }))
);

RLS でラップした関数の使用

// convex/messages.ts
import { queryWithRLS, mutationWithRLS } from "./functions";
import { v } from "convex/values";

// This query automatically enforces RLS rules
export const list = queryWithRLS({
  args: { conversationId: v.id("conversations") },
  returns: v.array(
    v.object({
      _id: v.id("messages"),
      _creationTime: v.number(),
      content: v.string(),
      authorId: v.string(),
    })
  ),
  handler: async (ctx, args) => {
    // RLS automatically filters out unauthorized messages
    return await ctx.db
      .query("messages")
      .withIndex("by_conversation", (q) =>
        q.eq("conversationId", args.conversationId)
      )
      .collect();
  },
});

// This mutation automatically enforces RLS rules
export const update = mutationWithRLS({
  args: { messageId: v.id("messages"), content: v.string() },
  returns: v.null(),
  handler: async (ctx, args) => {
    // RLS checks if user can modify this message
    await ctx.db.patch(args.messageId, { content: args.content });
    return null;
  },
});

リレーションシップヘルパー

手動で検索せずにリレーションシップを簡単にたどれます。

使用可能なヘルパー

import {
  getAll,
  getOneFrom,
  getManyFrom,
  getManyVia,
} from "convex-helpers/server/relationships";

1 対 1 のリレーションシップ

// Get single related document via back reference
const profile = await getOneFrom(
  ctx.db,
  "profiles", // target table
  "userId", // index field
  user._id // value to match
);

1 対多（ID 配列による）

// Load multiple documents by IDs
const users = await getAll(ctx.db, userIds);
// Returns array of documents in same order as IDs (null for missing)

1 対多（インデックス経由）

// Get all posts by author
const posts = await getManyFrom(
  ctx.db,
  "posts", // target table
  "by_authorId", // index name
  author._id // value to match
);

多対多（結合テーブル経由）

// Schema:
// posts: { title: v.string() }
// categories: { name: v.string() }
// postCategories: { postId: v.id("posts"), categoryId: v.id("categories") }
//   .index("by_post", ["postId"])
//   .index("by_category", ["categoryId"])

// Get all categories for a post
const categories = await getManyVia(
  ctx.db,
  "postCategories", // join table
  "categoryId", // field pointing to target
  "by_post", // index to query join table
  post._id // source ID
);

// Get all posts in a category
const posts = await getManyVia(
  ctx.db,
  "postCategories",
  "postId",
  "by_category",
  category._id
);

完全な例

// convex/posts.ts
import { query } from "./_generated/server";
import { v } from "convex/values";
import {
  getOneFrom,
  getManyFrom,
  getManyVia,
} from "convex-helpers/server/relationships";

export const getPostWithDetails = query({
  args: { postId: v.id("posts") },
  returns: v.union(
    v.object({
      post: v.object({
        _id: v.id("posts"),
        title: v.string(),
        body: v.string(),
      }),
      author: v.union(
        v.object({
          _id: v.id("users"),
          name: v.string(),
        }),
        v.null()
      ),
      comments: v.array(
        v.object({
          _id: v.id("comments"),
          body: v.string(),
        })
      ),
      categories: v.array(
        v.object({
          _id: v.id("categories"),
          name: v.string(),
        })
      ),
    }),
    v.null()
  ),
  handler: async (ctx, args) => {
    const post = await ctx.db.get(args.postId);
    if (!post) return null;

    const [author, comments, categories] = await Promise.all([
      // One-to-one: post -> author
      ctx.db.get(post.authorId),

      // One-to-many: post -> comments
      getManyFrom(ctx.db, "comments", "by_post", post._id),

      // Many-to-many: post -> categories (via join table)
      getManyVia(ctx.db, "postCategories", "categoryId", "by_post", post._id),
    ]);

    return {
      post: { _id: post._id, title: post.title, body: post.body },
      author: author ? { _id: author._id, name: author.name } : null,
      comments: comments.map((c) => ({ _id: c._id, body: c.body })),
      categories: categories
        .filter((c): c is NonNullable<typeof c> => c !== null)
        .map((c) => ({ _id: c._id, name: c.name })),
    };
  },
});

カスタム関数（認証ラッパー）

認証機能を組み込んだ、再利用可能な関数ラッパーを作成します。

基本的な認証ラッパー

// convex/functions.ts
import {
  customQuery,
  customMutation,
} from "convex-helpers/server/customFunctions";
import { query, mutation } from "./_generated/server";
import { Doc } from "./_generated/dataModel";

// Query that requires authentication
export const authedQuery = customQuery(query, {
  args: {},
  input: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    const user = await ctx.db
      .query("users")
      .withIndex("by_token", (q) =>
        q.eq("tokenIdentifier", identity.tokenIdentifier)
      )
      .unique();

    if (!user) throw new Error("User not found");

    return { ctx: { ...ctx, user }, args };
  },
});

// Mutation that requires authentication
export const authedMutation = customMutation(mutation, {
  args: {},
  input: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    const user = await ctx.db
      .query("users")
      .withIndex("by_token", (q) =>
        q.eq("tokenIdentifier", identity.tokenIdentifier)
      )
      .unique();

    if (!user) throw new Error("User not found");

    return { ctx: { ...ctx, user }, args };
  },
});

認証済み関数の使用

// convex/profile.ts
import { authedQuery, authedMutation } from "./functions";
import { v } from "convex/values";

// ctx.user is guaranteed to exist
export const getMyProfile = authedQuery({
  args: {},
  returns: v.object({
    _id: v.id("users"),
    name: v.string(),
    email: v.string(),
  }),
  handler: async (ctx) => {
    // ctx.user is typed and guaranteed to exist!
    return {
      _id: ctx.user._id,
      name: ctx.user.name,
      email: ctx.user.email,
    };
  },
});

export const updateMyName = authedMutation({
  args: { name: v.string() },
  returns: v.null(),
  handler: async (ctx, args) => {
    await ctx.db.patch(ctx.user._id, { name: args.name });
    return null;
  },
});

ロールベースの認証ラッパー

// convex/functions.ts
export const adminQuery = customQuery(query, {
  args: {},
  input: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    const user = await ctx.db
      .query("users")
      .withIndex("by_token", (q) =>
        q.eq("tokenIdentifier", identity.tokenIdentifier)
      )
      .unique();

    if (!user) throw new Error("User not found");
    if (user.role !== "admin") throw new Error("Admin access required");

    return { ctx: { ...ctx, user }, args };
  },
});

// Usage
export const listAllUsers = adminQuery({
  args: {},
  returns: v.array(
    v.object({
      _id: v.id("users"),
      name: v.string(),
      role: v.string(),
    })
  ),
  handler: async (ctx) => {
    const users = await ctx.db.query("users").collect();
    return users.map((u) => ({ _id: u._id, name: u.name, role: u.role }));
  },
});

Workpool（並行処理の制御）

Workpool は並列度の上限を設けて同時実行を管理します。次の用途に便利です。

• OCC の競合を避けるために書き込みを直列化する
• 上限付きで並列処理を分散実行する
• 外部 API 呼び出しにレート制限を設ける

Workpool の設定

まず、Workpool コンポーネントをインストールして設定します。

// convex/convex.config.ts
import { defineApp } from "convex/server";
import workpool from "@convex-dev/workpool/convex.config";

const app = defineApp();
app.use(workpool, { name: "workpool" });

export default app;

Workpool の使用

// convex/counters.ts
import { Workpool } from "@convex-dev/workpool";
import { components, internal } from "./_generated/api";
import { mutation, internalMutation } from "./_generated/server";
import { v } from "convex/values";

// Create workpool with parallelism limit
const counterPool = new Workpool(components.workpool, {
  maxParallelism: 1, // Serialize all counter updates
});

// Public mutation enqueues work
export const incrementCounter = mutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    await counterPool.enqueueMutation(ctx, internal.counters.doIncrement, {});
    return null;
  },
});

// Internal mutation does the actual work
export const doIncrement = internalMutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const counter = await ctx.db.query("counters").unique();
    if (counter) {
      await ctx.db.patch(counter._id, { count: counter.count + 1 });
    }
    return null;
  },
});

上限付きの並列処理

// Process many items with limited concurrency
const processingPool = new Workpool(components.workpool, {
  maxParallelism: 5, // Process 5 items at a time
});

export const processAll = mutation({
  args: { itemIds: v.array(v.id("items")) },
  returns: v.null(),
  handler: async (ctx, args) => {
    for (const itemId of args.itemIds) {
      await processingPool.enqueueAction(ctx, internal.items.processOne, {
        itemId,
      });
    }
    return null;
  },
});

レート制限

convex-helpers を使用して、アプリケーションレベルのレート制限を実装します。

レートリミッターの設定

// convex/rateLimit.ts
import { RateLimiter } from "convex-helpers/server/rateLimit";
import { components } from "./_generated/api";

export const rateLimiter = new RateLimiter(components.rateLimit, {
  // Global rate limit
  global: {
    kind: "token bucket",
    rate: 100, // 100 requests
    period: 60000, // per minute
  },

  // Per-user rate limit
  perUser: {
    kind: "token bucket",
    rate: 10,
    period: 60000,
  },
});

レートリミッターの使用

// convex/api.ts
import { mutation } from "./_generated/server";
import { v } from "convex/values";
import { rateLimiter } from "./rateLimit";

export const createPost = mutation({
  args: { title: v.string(), body: v.string() },
  returns: v.union(
    v.id("posts"),
    v.object({
      error: v.string(),
      retryAfter: v.number(),
    })
  ),
  handler: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    // Check rate limit
    const { ok, retryAfter } = await rateLimiter.limit(ctx, "perUser", {
      key: identity.subject,
    });

    if (!ok) {
      // Add jitter to prevent thundering herd
      const jitter = Math.random() * 1000;
      return {
        error: "Rate limit exceeded",
        retryAfter: retryAfter + jitter,
      };
    }

    const postId = await ctx.db.insert("posts", {
      title: args.title,
      body: args.body,
      authorId: identity.subject,
    });

    return postId;
  },
});

パターンの組み合わせ

トリガー + RLS + カスタム関数

// convex/functions.ts
import {
  mutation as rawMutation,
  query as rawQuery,
} from "./_generated/server";
import { Triggers } from "convex-helpers/server/triggers";
import {
  wrapDatabaseReader,
  wrapDatabaseWriter,
} from "convex-helpers/server/rowLevelSecurity";
import {
  customCtx,
  customQuery,
  customMutation,
} from "convex-helpers/server/customFunctions";

// Set up triggers
const triggers = new Triggers<DataModel>();
triggers.register("posts", async (ctx, change) => {
  if (change.operation === "insert") {
    // Update author's post count
    const author = await ctx.db.get(change.newDoc!.authorId);
    if (author) {
      await ctx.db.patch(author._id, {
        postCount: (author.postCount ?? 0) + 1,
      });
    }
  }
});

// Set up RLS rules
async function rlsRules(ctx: QueryCtx) {
  const identity = await ctx.auth.getUserIdentity();
  return {
    posts: {
      read: async () => true,
      modify: async (_, post) => post.authorId === identity?.subject,
    },
  };
}

// Combine everything into authenticated, RLS-protected, trigger-enabled functions
export const authedMutation = customMutation(rawMutation, {
  args: {},
  input: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    const user = await ctx.db
      .query("users")
      .withIndex("by_token", (q) =>
        q.eq("tokenIdentifier", identity.tokenIdentifier)
      )
      .unique();

    if (!user) throw new Error("User not found");

    // Wrap DB with triggers and RLS
    const wrappedDb = wrapDatabaseWriter(
      ctx,
      triggers.wrapDB(ctx).db,
      await rlsRules(ctx)
    );

    return { ctx: { ...ctx, user, db: wrappedDb }, args };
  },
});

よくある落とし穴

落とし穴 1：トリガーによる OCC の競合

❌ 誤り：

// This trigger updates a single global counter - will cause OCC under load
triggers.register("posts", async (ctx, change) => {
  if (change.operation === "insert") {
    const stats = await ctx.db.query("globalStats").unique();
    await ctx.db.patch(stats!._id, { postCount: stats!.postCount + 1 });
  }
});

✅ 正しい例：

// Use sharding or Workpool for high-contention updates
triggers.register("posts", async (ctx, change) => {
  if (change.operation === "insert") {
    const shardId = Math.floor(Math.random() * 10);
    await ctx.db.insert("postCountShards", { shardId, delta: 1 });
  }
});

落とし穴 2：RLS ルールにテーブルが含まれていない

❌ 誤り：

// Missing rules for some tables - they'll be unprotected!
async function rlsRules(ctx: QueryCtx) {
  return {
    users: { read: async () => true, modify: async () => false },
    // Missing posts, messages, etc.!
  };
}

✅ 正しい例：

// Define rules for ALL tables
async function rlsRules(ctx: QueryCtx) {
  return {
    users: { read: async () => true, modify: async () => false },
    posts: { read: async () => true, modify: async () => true },
    messages: { read: async () => true, modify: async () => true },
    // ... all other tables
  } satisfies Rules<QueryCtx, DataModel>;
}

クイックリファレンス

インポートパターン

// Triggers
import { Triggers } from "convex-helpers/server/triggers";

// RLS
import {
  Rules,
  wrapDatabaseReader,
  wrapDatabaseWriter,
} from "convex-helpers/server/rowLevelSecurity";

// Custom Functions
import {
  customCtx,
  customQuery,
  customMutation,
} from "convex-helpers/server/customFunctions";

// Relationships
import {
  getAll,
  getOneFrom,
  getManyFrom,
  getManyVia,
} from "convex-helpers/server/relationships";

// Workpool
import { Workpool } from "@convex-dev/workpool";

// Rate Limiter
import { RateLimiter } from "convex-helpers/server/rateLimit";