ddd refactor: project-members

apps/rowboat/app/actions/project_actions.ts

@@ -1,7 +1,7 @@
 'use server';
 import { redirect } from "next/navigation";
 import { ObjectId } from "mongodb";
-import { db, dataSourcesCollection, projectsCollection, projectMembersCollection, dataSourceDocsCollection } from "../lib/mongodb";
+import { db, dataSourcesCollection, projectsCollection, dataSourceDocsCollection } from "../lib/mongodb";
 import { z } from 'zod';
 import crypto from 'crypto';
 import { revalidatePath } from "next/cache";
@@ -19,6 +19,7 @@ import { ICreateApiKeyController } from "@/src/interface-adapters/controllers/ap
 import { IListApiKeysController } from "@/src/interface-adapters/controllers/api-keys/list-api-keys.controller";
 import { IDeleteApiKeyController } from "@/src/interface-adapters/controllers/api-keys/delete-api-key.controller";
 import { IApiKeysRepository } from "@/src/application/repositories/api-keys.repository.interface";
+import { IProjectMembersRepository } from "@/src/application/repositories/project-members.repository.interface";
 const KLAVIS_API_KEY = process.env.KLAVIS_API_KEY || '';
 
 const projectActionAuthorizationPolicy = container.resolve<IProjectActionAuthorizationPolicy>('projectActionAuthorizationPolicy');
@@ -26,6 +27,7 @@ const createApiKeyController = container.resolve<ICreateApiKeyController>('creat
 const listApiKeysController = container.resolve<IListApiKeysController>('listApiKeysController');
 const deleteApiKeyController = container.resolve<IDeleteApiKeyController>('deleteApiKeyController');
 const apiKeysRepository = container.resolve<IApiKeysRepository>('apiKeysRepository');
+const projectMembersRepository = container.resolve<IProjectMembersRepository>('projectMembersRepository');
 
 export async function listTemplates() {
     const templatesArray = Object.entries(templates)
@@ -94,11 +96,9 @@ async function createBaseProject(
     });
 
     // Add user to project
-    await projectMembersCollection.insertOne({
+    await projectMembersRepository.create({
         userId: user._id,
-        projectId: projectId,
-        createdAt: (new Date()).toISOString(),
-        lastUpdatedAt: (new Date()).toISOString(),
+        projectId,
     });
 
     // Add first api key
@@ -159,9 +159,13 @@ export async function getProjectConfig(projectId: string): Promise<WithStringId<
 
 export async function listProjects(): Promise<z.infer<typeof Project>[]> {
     const user = await authCheck();
-    const memberships = await projectMembersCollection.find({
-        userId: user._id,
-    }).toArray();
+    const memberships = [];
+    let cursor = undefined;
+    do {
+        const result = await projectMembersRepository.findByUserId(user._id, cursor);
+        memberships.push(...result.items);
+        cursor = result.nextCursor;
+    } while (cursor);
     const projectIds = memberships.map((m) => m.projectId);
     const projects = await projectsCollection.find({
         _id: { $in: projectIds },
@@ -251,9 +255,7 @@ export async function deleteProject(projectId: string) {
     });
 
     // delete project members
-    await projectMembersCollection.deleteMany({
-        projectId,
-    });
+    await projectMembersRepository.deleteByProjectId(projectId);
 
     // delete workflow versions
     await db.collection('agent_workflows').deleteMany({

apps/rowboat/app/lib/auth.ts

@@ -1,6 +1,6 @@
 import { z } from "zod";
 import { ObjectId } from "mongodb";
-import { usersCollection, projectsCollection, projectMembersCollection } from "./mongodb";
+import { usersCollection } from "./mongodb";
 import { auth0 } from "./auth0";
 import { User, WithStringId } from "./types/types";
 import { USE_AUTH } from "./feature_flags";
@@ -59,11 +59,6 @@ export async function requireAuth(): Promise<WithStringId<z.infer<typeof User>>>
         console.log(`creating new user id ${doc._id.toString()} for session id ${user.sub}`);
         await usersCollection.insertOne(doc);
 
-        // since auth feature was rolled out later,
-        // set all project authors to new user id instead
-        // of user.sub
-        await updateProjectRefs(user.sub, doc._id.toString());
-
         dbUser = {
             ...doc,
             _id: doc._id.toString(),
@@ -77,25 +72,6 @@ export async function requireAuth(): Promise<WithStringId<z.infer<typeof User>>>
     };
 }
 
-async function updateProjectRefs(sessionUserId: string, dbUserId: string) {
-    await projectsCollection.updateMany({
-        createdByUserId: sessionUserId
-    }, {
-        $set: {
-            createdByUserId: dbUserId,
-            lastUpdatedAt: new Date().toISOString(),
-        }
-    });
-
-    await projectMembersCollection.updateMany({
-        userId: sessionUserId
-    }, {
-        $set: {
-            userId: dbUserId,
-        }
-    });
-}
-
 export async function getUserFromSessionId(sessionUserId: string): Promise<WithStringId<z.infer<typeof User>> | null> {
     if (!USE_AUTH) {
         return GUEST_DB_USER;

apps/rowboat/app/lib/mongodb.ts

@@ -2,7 +2,7 @@ import { MongoClient } from "mongodb";
 import { User, Webpage } from "./types/types";
 import { Workflow } from "./types/workflow_types";
 import { ApiKey } from "@/src/entities/models/api-key";
-import { ProjectMember } from "./types/project_types";
+import { ProjectMember } from "@/src/entities/models/project-member";
 import { Project } from "./types/project_types";
 import { EmbeddingDoc } from "./types/datasource_types";
 import { DataSourceDoc } from "./types/datasource_types";
@@ -17,7 +17,6 @@ export const db = client.db("rowboat");
 export const dataSourcesCollection = db.collection<z.infer<typeof DataSource>>("sources");
 export const dataSourceDocsCollection = db.collection<z.infer<typeof DataSourceDoc>>("source_docs");
 export const projectsCollection = db.collection<z.infer<typeof Project>>("projects");
-export const projectMembersCollection = db.collection<z.infer<typeof ProjectMember>>("project_members");
 export const agentWorkflowsCollection = db.collection<z.infer<typeof Workflow>>("agent_workflows");
 export const chatsCollection = db.collection<z.infer<typeof apiV1.Chat>>("chats");
 export const chatMessagesCollection = db.collection<z.infer<typeof apiV1.ChatMessage>>("chat_messages");

apps/rowboat/app/lib/types/project_types.ts

@@ -35,10 +35,3 @@ export const Project = z.object({
     composioConnectedAccounts: z.record(z.string(), ComposioConnectedAccount).optional(),
     customMcpServers: z.record(z.string(), CustomMcpServer).optional(),
 });
-
-export const ProjectMember = z.object({
-    userId: z.string(),
-    projectId: z.string(),
-    createdAt: z.string().datetime(),
-    lastUpdatedAt: z.string().datetime(),
-});

apps/rowboat/src/application/policies/project-action-authorization.policy.ts

@@ -36,7 +36,7 @@ export class ProjectActionAuthorizationPolicy implements IProjectActionAuthoriza
             if (!userId) {
                 throw new BadRequestError('User ID is required');
             }
-            const membership = await this.projectMembersRepository.checkMembership(projectId, userId);
+            const membership = await this.projectMembersRepository.exists(projectId, userId);
             if (!membership) {
                 throw new NotAuthorizedError('User is not a member of the project');
             }

apps/rowboat/src/application/repositories/project-members.repository.interface.ts

@@ -1,3 +1,39 @@
+import { ProjectMember } from "@/src/entities/models/project-member";
+import { PaginatedList } from "@/src/entities/common/paginated-list";
+import { z } from "zod";
+
+export const CreateProjectMemberSchema = ProjectMember.pick({
+    userId: true,
+    projectId: true,
+});
+
 export interface IProjectMembersRepository {
-    checkMembership(projectId: string, userId: string): Promise<boolean>;
+    /**
+     * Creates a new project member association. If the association already exists, returns the existing member.
+     * @param data - The data required to create a project member (userId and projectId).
+     * @returns A promise that resolves to the created or existing ProjectMember object.
+     */
+    create(data: z.infer<typeof CreateProjectMemberSchema>): Promise<z.infer<typeof ProjectMember>>;
+
+    /**
+     * Finds all project memberships for a given user, returned as a paginated list.
+     * @param userId - The ID of the user whose project memberships are to be retrieved.
+     * @returns A promise that resolves to a paginated list of ProjectMember objects.
+     */
+    findByUserId(userId: string, cursor?: string, limit?: number): Promise<z.infer<ReturnType<typeof PaginatedList<typeof ProjectMember>>>>;
+
+    /**
+     * Deletes all project member associations for a given project.
+     * @param projectId - The ID of the project whose member associations should be deleted.
+     * @returns A promise that resolves when the operation is complete.
+     */
+    deleteByProjectId(projectId: string): Promise<void>;
+
+    /**
+     * Checks if a specific membership exists.
+     * @param projectId - The ID of the project.
+     * @param userId - The ID of the user.
+     * @returns A promise that resolves to true if the user is a member of the project, false otherwise.
+     */
+    exists(projectId: string, userId: string): Promise<boolean>;
 }

apps/rowboat/src/entities/models/project-member.ts

@@ -0,0 +1,9 @@
+import { z } from "zod";
+
+export const ProjectMember = z.object({
+    id: z.string(),
+    userId: z.string(),
+    projectId: z.string(),
+    createdAt: z.string().datetime(),
+    lastUpdatedAt: z.string().datetime(),
+});

apps/rowboat/src/infrastructure/repositories/mongodb.project-members.repository.ts

@@ -1,12 +1,86 @@
-import { IProjectMembersRepository } from "@/src/application/repositories/project-members.repository.interface";
-import { projectMembersCollection } from "@/app/lib/mongodb";
+import { CreateProjectMemberSchema, IProjectMembersRepository } from "@/src/application/repositories/project-members.repository.interface";
+import { ProjectMember } from "@/src/entities/models/project-member";
+import { db } from "@/app/lib/mongodb";
+import { z } from "zod";
+import { ObjectId } from "mongodb";
+import { PaginatedList } from "@/src/entities/common/paginated-list";
+
+const docSchema = ProjectMember.omit({
+    id: true,
+});
 
 export class MongoDBProjectMembersRepository implements IProjectMembersRepository {
-    async checkMembership(projectId: string, userId: string): Promise<boolean> {
-        const membership = await projectMembersCollection.findOne({
+    private collection = db.collection<z.infer<typeof docSchema>>('project_members');
+
+    async create(data: z.infer<typeof CreateProjectMemberSchema>): Promise<z.infer<typeof ProjectMember>> {
+        // this has to be an upsert operation
+        const result = await this.collection.findOneAndUpdate(
+            {
+                userId: data.userId,
+                projectId: data.projectId,
+            },
+            {
+                $set: {
+                    ...data,
+                    createdAt: new Date().toISOString(),
+                    lastUpdatedAt: new Date().toISOString(),
+                }
+            },
+            {
+                upsert: true,
+                returnDocument: 'after',
+            }
+        );
+
+        if (!result) {
+            throw new Error('Failed to create project member');
+        }
+
+        const { _id, ...rest } = result;
+
+        return {
+            ...rest,
+            id: _id.toString(),
+        };
+    }
+
+    async findByUserId(userId: string, cursor?: string, limit: number = 50): Promise<z.infer<ReturnType<typeof PaginatedList<typeof ProjectMember>>>> {
+        const query: any = { userId };
+
+        if (cursor) {
+            query._id = { $lt: new ObjectId(cursor) };
+        }
+
+        const results = await this.collection
+            .find(query)
+            .sort({ _id: -1 })
+            .limit(limit + 1) // Fetch one extra to determine if there's a next page
+            .toArray();
+
+        const hasNextPage = results.length > limit;
+        const items = results.slice(0, limit).map(doc => {
+            const { _id, ...rest } = doc;
+            return {
+                ...rest,
+                id: _id.toString(),
+            };
+        });
+
+        return {
+            items,
+            nextCursor: hasNextPage ? results[limit - 1]._id.toString() : null,
+        };
+    }
+
+    async exists(projectId: string, userId: string): Promise<boolean> {
+        const membership = await this.collection.findOne({
             projectId,
             userId,
         });
         return !!membership;
     }
+
+    async deleteByProjectId(projectId: string): Promise<void> {
+        await this.collection.deleteMany({ projectId });
+    }
 }