rewrote get-merge-requests (#98)

* rewrote get-merge-requests

* Added message for every page of MRs

* fixed typings

* removed unused imports

* internal IDs added

* publishing events

---------

Co-authored-by: Ante Koceić <ante@crocoder.dev>
Co-authored-by: David Abram <david@crocoder.dev>

Author: 3 people

apps/extract-stack/src/events.ts

@@ -17,14 +17,18 @@ const metadataSchema = z.object({
   userId: z.string(),
 });
 const extractMergeRequestEventSchema = z.object({
-  mergeRequests: z.array(MergeRequestSchema),
+  mergeRequestIds: z.array(MergeRequestSchema.shape.id),
 });
 
-export const extractMergeRequestsEvent = {
-  schemaShape: extractMergeRequestEventSchema.shape,
+export type extractMergeRequestsEventMessage = z.infer<typeof extractMergeRequestEventSchema>;
+
+export const extractMergeRequestsEvent = createEvent({
   source: "extract",
-  detailType: "mergeRequest",
-};
+  type: "mergeRequest",
+  propertiesShape: extractMergeRequestEventSchema.shape,
+  eventBusName: EventBus.ExtractBus.eventBusName,
+  metadataShape: metadataSchema.shape,
+});
 
 export const extractRepositoryEvent = createEvent({
   source: "extract",

apps/extract-stack/src/extract-merge-requests.ts

@@ -6,17 +6,18 @@ import { EventHandler } from "sst/node/event-bus";
 
 import {
   getMergeRequests,
-  getPaginationData,
   type Context,
   type GetMergeRequestsEntities,
   type GetMergeRequestsSourceControl,
 } from "@acme/extract-functions";
-import { mergeRequests } from "@acme/extract-schema";
+import { mergeRequests, namespaces, repositories } from "@acme/extract-schema";
 import { GitHubSourceControl, GitlabSourceControl } from "@acme/source-control";
 
-import { extractRepositoryEvent } from "./events";
+import { extractMergeRequestsEvent, extractRepositoryEvent } from "./events";
 import { extractMergeRequestMessage } from "./messages";
+import type { extractRepositoryData } from "./messages";
 import { QueueHandler } from "./create-message";
+import { eq } from "drizzle-orm";
 
 const clerkClient = Clerk({ secretKey: Config.CLERK_SECRET_KEY });
 const client = createClient({
@@ -58,38 +59,57 @@ const initSourceControl = async (userId: string, sourceControl: 'github' | 'gitl
 }
 
 export const eventHandler = EventHandler(extractRepositoryEvent, async (evt) => {
+  if (!evt.properties.namespaceId) throw new Error("Missing namespaceId");
+
+  const repository = await db.select().from(repositories).where(eq(repositories.id, evt.properties.repositoryId)).get();
+  const namespace = await db.select().from(namespaces).where(eq(namespaces.id, evt.properties.namespaceId)).get();
+
+  if (!repository) throw new Error("invalid repo id");
+  if (!namespace) throw new Error("Invalid namespace id");
 
-  const externalRepositoryId = evt.properties.repository.externalId;
-  const repositoryName = evt.properties.repository.name;
-  const namespace = evt.properties.namespace;
-  const repository = evt.properties.repository;
   const sourceControl = evt.metadata.sourceControl;
-  const repositoryId = evt.properties.repository.id;
 
   context.integrations.sourceControl = await initSourceControl(evt.metadata.userId, sourceControl)
 
-  const { paginationInfo } = await getPaginationData(
-    {
-      externalRepositoryId: externalRepositoryId,
-      namespaceName: namespace?.name || "",
-      repositoryName: repositoryName,
-      repositoryId: repositoryId,
-    },
-    context,
-  );
+    const { mergeRequests, paginationInfo } = await getMergeRequests(
+      {
+        externalRepositoryId: repository.externalId,
+        namespaceName: namespace?.name || "",
+        repositoryName: repository.name,
+        repositoryId: repository.id,
+        perPage: 10,
+      }, context,
+    );
+
+    await extractMergeRequestsEvent.publish({mergeRequestIds: mergeRequests.map(mr => mr.id)}, {
+      version: 1,
+      caller: 'extract-merge-requests',
+      sourceControl,
+      userId: evt.metadata.userId,
+      timestamp: new Date().getTime(),
+    });
+
+    const arrayOfExtractMergeRequests: extractRepositoryData[] = [];
+    for(let i = 2; i <= paginationInfo.totalPages; i++ ) {
+      arrayOfExtractMergeRequests.push({
+        repository,
+        namespace: namespace,
+        pagination: {
+          page: i,
+          perPage: paginationInfo.perPage,
+          totalPages: paginationInfo.totalPages
+        }
+      });
+    }
+     
+    await extractMergeRequestMessage.sendAll(arrayOfExtractMergeRequests, { 
+      version: 1,
+      caller: 'extract-merge-requests',
+      sourceControl,
+      userId: evt.metadata.userId,
+      timestamp: new Date().getTime(),
+    });
 
-  for (let index = 1; index <= paginationInfo.totalPages; index++) {
-
-    await extractMergeRequestMessage.send({
-      repository,
-      namespace,
-      pagination: {
-        page: index,
-        perPage: paginationInfo.perPage,
-        totalPages: paginationInfo.totalPages
-      }
-    }, { caller: 'extract-merge-requests', timestamp: new Date().getTime(), version: 1, sourceControl, userId: evt.metadata.userId });
-  }
 });
 
 export const queueHandler = QueueHandler(extractMergeRequestMessage, async (message) => {
@@ -103,8 +123,7 @@ export const queueHandler = QueueHandler(extractMergeRequestMessage, async (mess
 
   const {namespace, pagination, repository} = message.content;
 
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars  
-  const { mergeRequests } = await getMergeRequests(
+  const {mergeRequests} = await getMergeRequests(
     {
       externalRepositoryId: repository.externalId,
       namespaceName: namespace?.name || "",
@@ -115,4 +134,13 @@ export const queueHandler = QueueHandler(extractMergeRequestMessage, async (mess
     },
     context,
   );
+
+  await extractMergeRequestsEvent.publish({mergeRequestIds: mergeRequests.map(mr => mr.id)}, {
+    version: 1,
+    caller: 'extract-merge-requests',
+    sourceControl: message.metadata.sourceControl,
+    userId: message.metadata.userId,
+    timestamp: new Date().getTime(),
+  });
+
 })

apps/extract-stack/src/messages.ts

@@ -16,6 +16,8 @@ const extractRepositoryDataSchema = z.object({
   pagination: paginationSchema
 });
 
+export type extractRepositoryData = z.infer<typeof extractRepositoryDataSchema>;
+
 const metadataSchema = z.object({
   version: z.number(),
   timestamp: z.number(),
@@ -34,4 +36,4 @@ export const extractMergeRequestMessage = createMessage({
   metadataShape: metadataSchema.shape,
   contentShape: extractRepositoryDataSchema.shape,
   queueUrl: Queue.MRQueue.queueUrl
-});
+});

apps/extract-stack/src/stack.ts

@@ -70,6 +70,12 @@ export function ExtractStack({ stack }: StackContext) {
   });
 
   mergeRequestQueue.addConsumer(stack, {
+    cdk: {
+      eventSource: {
+        batchSize: 1,
+        maxConcurrency: 20,
+      },
+    },
     function: {
       bind: [
         bus,

packages/functions/extract/src/get-merge-requests.ts

@@ -1,6 +1,8 @@
 import type { MergeRequest } from "@acme/extract-schema";
 import type { ExtractFunction, Entities } from "./config";
 import type { Pagination, SourceControl } from "@acme/source-control";
+import type { LibSQLDatabase } from "drizzle-orm/libsql";
+import type { BetterSQLite3Database } from "drizzle-orm/better-sqlite3";
 
 export type GetMergeRequestsInput = {
   externalRepositoryId: number;
@@ -25,28 +27,6 @@ export type GetMergeRequestsEntities = Pick<Entities, "mergeRequests">;
 
 export type GetMergeRequestsFunction = ExtractFunction<GetMergeRequestsInput, GetMergeRequestsOutput, GetMergeRequestsSourceControl, GetMergeRequestsEntities>;
 
-export type GetPaginationDataFunction = ExtractFunction<GetMergeRequestsInput, GetPaginationDataOutput, GetMergeRequestsSourceControl, GetMergeRequestsEntities>;
-
-export const getPaginationData: GetPaginationDataFunction = async (
-  { externalRepositoryId, namespaceName, repositoryName, repositoryId },
-  { integrations },
-) => {
-  if (!integrations.sourceControl) {
-    throw new Error("Source control integration not configured");
-  }
-
-  const { pagination } = await integrations.sourceControl.fetchMergeRequests(
-    externalRepositoryId,
-    namespaceName,
-    repositoryName,
-    repositoryId,
-    {},
-  );
-
-  return {
-    paginationInfo: pagination,
-  };
-};
 
 export const getMergeRequests: GetMergeRequestsFunction = async (
   { externalRepositoryId, namespaceName, repositoryName, repositoryId, page, perPage},
@@ -59,12 +39,17 @@ export const getMergeRequests: GetMergeRequestsFunction = async (
 
   const { mergeRequests, pagination } = await integrations.sourceControl.fetchMergeRequests(externalRepositoryId, namespaceName, repositoryName, repositoryId, {}, page, perPage);
 
-    const insertedMergeRequests = await db.insert(entities.mergeRequests).values(mergeRequests)
-      .onConflictDoNothing({ target: entities.mergeRequests.externalId }).returning()
-      .all();
+  const insertedMergeRequests = await (db as (LibSQLDatabase & BetterSQLite3Database)).transaction(async (tx) => {
+    return Promise.all(mergeRequests.map(mergeRequest =>
+      tx.insert(entities.mergeRequests).values(mergeRequest)
+        .onConflictDoUpdate({ target: entities.mergeRequests.externalId, set: { updatedAt: new Date() } })
+        .returning()
+        .get()
+    ));
+  });
 
     return {
       mergeRequests: insertedMergeRequests,
       paginationInfo: pagination,
     };
-  };
+  };

packages/integrations/source-control/src/gitlab/index.ts

@@ -57,7 +57,7 @@ export class GitlabSourceControl implements SourceControl {
   async fetchMergeRequests(externalRepositoryId: number, namespaceName = '', repositoryName = '', repositoryId: number, creationPeriod: TimePeriod = {}, page?: number, perPage?: number): Promise<{ mergeRequests: NewMergeRequest[], pagination: Pagination }> {
     const { data, paginationInfo } = await this.api.MergeRequests.all({
       projectId: externalRepositoryId,
-      page,
+      page: page || 1,
       perPage,
       pagination: 'offset',
       showExpanded: true,