implement perpetual GC for notion

dust-tt · Jan 24, 2024 · 1c279ef · 1c279ef
1 parent e7062f2
commit 1c279ef
Show file tree

Hide file tree

Showing 3 changed files with 94 additions and 10 deletions.
diff --git a/connectors/src/connectors/notion/temporal/activities.ts b/connectors/src/connectors/notion/temporal/activities.ts
@@ -970,11 +970,13 @@ export async function cachePage({
   connectorId,
   pageId,
   runTimestamp,
+  topLevelWorkflowId,
   loggerArgs,
 }: {
   connectorId: ModelId;
   pageId: string;
   runTimestamp: number;
+  topLevelWorkflowId: string;
   loggerArgs: Record<string, string | number>;
 }): Promise<{
   skipped: boolean;
@@ -1031,6 +1033,7 @@ export async function cachePage({
     where: {
       notionPageId: pageId,
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
   if (pageInCache) {
@@ -1078,6 +1081,7 @@ export async function cachePage({
     createdTime: notionPage.created_time,
     lastEditedTime: notionPage.last_edited_time,
     url: notionPage.url,
+    workflowId: topLevelWorkflowId,
   });
 
   return {
@@ -1092,13 +1096,15 @@ export async function cacheBlockChildren({
   cursor,
   currentIndexInParent,
   loggerArgs,
+  topLevelWorkflowId,
 }: {
   connectorId: ModelId;
   pageId: string;
   blockId: string | null;
   cursor: string | null;
   currentIndexInParent: number;
   loggerArgs: Record<string, string | number>;
+  topLevelWorkflowId: string;
 }): Promise<{
   nextCursor: string | null;
   blocksWithChildren: string[];
@@ -1181,6 +1187,7 @@ export async function cacheBlockChildren({
         notionPageId: pageId,
         notionBlockId: parsedBlocks.map((b) => b.id),
         connectorId: connector.id,
+        workflowId: topLevelWorkflowId,
       },
       attributes: ["notionBlockId"],
     });
@@ -1229,6 +1236,7 @@ export async function cacheBlockChildren({
         indexInParent: currentIndexInParent + i,
         childDatabaseTitle: block.childDatabaseTitle,
         connectorId: connector.id,
+        workflowId: topLevelWorkflowId,
       })
     )
   );
@@ -1245,11 +1253,13 @@ export async function cacheDatabaseChildren({
   connectorId,
   databaseId,
   cursor,
+  topLevelWorkflowId,
   loggerArgs,
 }: {
   connectorId: ModelId;
   databaseId: string;
   cursor: string | null;
+  topLevelWorkflowId: string;
   loggerArgs: Record<string, string | number>;
 }): Promise<{
   nextCursor: string | null;
@@ -1321,6 +1331,7 @@ export async function cacheDatabaseChildren({
     where: {
       notionPageId: resultPage.results.map((r) => r.id),
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
     attributes: ["notionPageId"],
   });
@@ -1356,6 +1367,7 @@ export async function cacheDatabaseChildren({
         createdTime: page.created_time,
         lastEditedTime: page.last_edited_time,
         url: page.url,
+        workflowId: topLevelWorkflowId,
       })
     )
   );
@@ -1501,12 +1513,14 @@ export async function renderAndUpsertPageFromCache({
   loggerArgs,
   runTimestamp,
   isFullSync,
+  topLevelWorkflowId,
 }: {
   connectorId: ModelId;
   pageId: string;
   loggerArgs: Record<string, string | number>;
   runTimestamp: number;
   isFullSync: boolean;
+  topLevelWorkflowId: string;
 }) {
   const connector = await Connector.findOne({
     where: {
@@ -1550,6 +1564,7 @@ export async function renderAndUpsertPageFromCache({
     where: {
       notionPageId: pageId,
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
   if (!pageCacheEntry) {
@@ -1563,6 +1578,7 @@ export async function renderAndUpsertPageFromCache({
     where: {
       notionPageId: pageId,
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
 
@@ -1594,6 +1610,7 @@ export async function renderAndUpsertPageFromCache({
     where: {
       parentId: Object.keys(blocksByParentId),
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
   const childDatabases: Record<string, NotionConnectorPageCacheEntry[]> = {};
@@ -1744,6 +1761,7 @@ export async function renderAndUpsertPageFromCache({
         notionId,
         connectorId: connector.id,
         resourceType: parentType,
+        workflowId: topLevelWorkflowId,
       });
     }
   }
@@ -1891,19 +1909,27 @@ export async function renderAndUpsertPageFromCache({
         notionId: id,
         resourceType: "database",
         connectorId: connector.id,
+        workflowId: topLevelWorkflowId,
       })
     ),
     ...pageEntriesToCreate.map((id) =>
       NotionConnectorResourcesToCheckCacheEntry.upsert({
         notionId: id,
         resourceType: "page",
         connectorId: connector.id,
+        workflowId: topLevelWorkflowId,
       })
     ),
   ]);
 }
 
-export async function clearConnectorCache(connectorId: ModelId) {
+export async function clearConnectorCache({
+  connectorId,
+  topLevelWorkflowId,
+}: {
+  connectorId: ModelId;
+  topLevelWorkflowId: string;
+}) {
   const connector = await Connector.findOne({
     where: {
       type: "notion",
@@ -1923,16 +1949,19 @@ export async function clearConnectorCache(connectorId: ModelId) {
   await NotionConnectorPageCacheEntry.destroy({
     where: {
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
   await NotionConnectorBlockCacheEntry.destroy({
     where: {
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
   await NotionConnectorResourcesToCheckCacheEntry.destroy({
     where: {
       connectorId: connector.id,
+      workflowId: topLevelWorkflowId,
     },
   });
 }