chore: more frequent etls (#394)

* don't publish empty merge-request extract events

* smaller

* oopsie

Author: dejan-crocoder

apps/stack/src/config/time-period.ts

@@ -0,0 +1,9 @@
+export const timePeriodOf = (timestamp: number, periodDuration: number, periodStartMargin: number, periodLatency: number) => {
+  const to = timestamp - (timestamp % periodDuration);
+  const from = to - (periodDuration + periodStartMargin);
+
+  return {
+    from: new Date(from - periodLatency),
+    to: new Date(to - periodLatency)
+  }
+}

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

@@ -64,6 +64,8 @@ export const mergeRequestSenderHandler = createMessageHandler({
       { ...context, db: getTenantDb(message.metadata.tenantId) },
     );
 
+    if (mergeRequests.length === 0) return;
+
     await extractMergeRequestsEvent.publish(
       {
         mergeRequestIds: mergeRequests.map((mr) => mr.id),
@@ -154,6 +156,8 @@ export const eventHandler = EventHandler(
       { ...context, db },
     );
 
+    if (mergeRequests.length === 0 && (paginationInfo.totalPages - paginationInfo.page) === 0) return;
+
     await insertEvent(
       {
         crawlId: ev.metadata.crawlId,
@@ -166,24 +170,26 @@ export const eventHandler = EventHandler(
       { db, entities: { events } },
     );
 
-    await extractMergeRequestsEvent.publish(
-      {
-        mergeRequestIds: mergeRequests.map((mr) => mr.id),
-        namespaceId: namespace.id,
-        repositoryId: repository.id,
-      },
-      {
-        crawlId: ev.metadata.crawlId,
-        version: 1,
-        caller: "extract-merge-requests",
-        sourceControl,
-        userId: ev.metadata.userId,
-        timestamp: new Date().getTime(),
-        from: ev.metadata.from,
-        to: ev.metadata.to,
-        tenantId: ev.metadata.tenantId,
-      },
-    );
+    if (mergeRequests.length !== 0) {
+      await extractMergeRequestsEvent.publish(
+        {
+          mergeRequestIds: mergeRequests.map((mr) => mr.id),
+          namespaceId: namespace.id,
+          repositoryId: repository.id,
+        },
+        {
+          crawlId: ev.metadata.crawlId,
+          version: 1,
+          caller: "extract-merge-requests",
+          sourceControl,
+          userId: ev.metadata.userId,
+          timestamp: new Date().getTime(),
+          from: ev.metadata.from,
+          to: ev.metadata.to,
+          tenantId: ev.metadata.tenantId,
+        },
+      );
+    }
 
     const arrayOfExtractMergeRequests = [];
     for (let i = paginationInfo.page + 1; i <= paginationInfo.totalPages; i++) {

apps/stack/src/extract/extract-tenants.ts

@@ -7,6 +7,7 @@ import { namespaces, repositories } from "@acme/extract-schema";
 import { eq } from "drizzle-orm";
 import { repositorySenderHandler } from "./extract-repository";
 import { ApiHandler, useJsonBody } from "sst/node/api";
+import { timePeriodOf } from "@stack/config/time-period";
 
 export const tenantSenderHandler = createMessageHandler({
   queueId: 'ExtractQueue',
@@ -62,18 +63,20 @@ export const cronHandler = async ()=> {
   const tenants = getTenants();
   const tenantIds = tenants.map(tenant => ({ tenantId: tenant.id }));
 
-  const utcTodayAt10AM = new Date();
-  utcTodayAt10AM.setUTCHours(10, 0, 0, 0);
-  const utcYesterdayAt10AM = new Date(utcTodayAt10AM);
-  utcYesterdayAt10AM.setHours(utcTodayAt10AM.getUTCHours() - 24);
+  const PERIOD_DURATION = 15 * 60 * 1000; // 15 minutes
+  const PERIOD_START_MARGIN = 5 * 60 * 1000; // 5 minutes
+  const PERIOD_LATENCY = 8 * 60 * 1000; // extract delay
+  const { from, to } = timePeriodOf(Date.now(), PERIOD_DURATION, PERIOD_START_MARGIN, -PERIOD_LATENCY);
 
+  console.log(`CRON-EXTRACT:`, { from, to, now: new Date() })
+  
   await sender.sendAll(tenantIds, {
     version: 1,
     caller: 'extract-tenant:cronHandler',
     timestamp: Date.now(),
     userId: CRON_USER_ID,
-    from: utcYesterdayAt10AM,
-    to: utcTodayAt10AM,
+    from,
+    to,
     tenantId: -1, // -1 means no db access ?
   });
 

apps/stack/src/transform/transform-tenant.ts

@@ -7,6 +7,7 @@ import { createMessageHandler } from "@stack/config/create-message";
 import { getTenantDb } from "@stack/config/get-tenant-db";
 import { and, eq, gt, lt } from "drizzle-orm";
 import { timelineSenderHandler } from "./transform-timeline";
+import { timePeriodOf } from "@stack/config/time-period";
 
 export const tenantSenderHandler = createMessageHandler({
   queueId: 'TransformQueue',
@@ -58,17 +59,19 @@ const { sender } = tenantSenderHandler;
 export const cronHandler = async () => {
   const tenants = getTenants().map(tenant => ({ tenantId: tenant.id }));
 
-  const utcTodayAt10AM = new Date();
-  utcTodayAt10AM.setUTCHours(10, 0, 0, 0);
-  const utcYesterdayAt10AM = new Date(utcTodayAt10AM);
-  utcYesterdayAt10AM.setHours(utcTodayAt10AM.getUTCHours() - 24);
+  const PERIOD_DURATION = 15 * 60 * 1000; // 15 minutes
+  const PERIOD_START_MARGIN = 5 * 60 * 1000; // 5 minutes
+  const PERIOD_LATENCY = (15 - 8) * 60 * 1000; // extract delay
+  const { from, to } = timePeriodOf(Date.now(), PERIOD_DURATION, PERIOD_START_MARGIN, PERIOD_LATENCY);
+
+  console.log(`CRON-TRANSFORM:`, { from, to, now: new Date() })
 
   await sender.sendAll(tenants, {
     version: 1,
     caller: 'transform-tenants:cronHandler',
     timestamp: Date.now(),
-    from: utcYesterdayAt10AM,
-    to: utcTodayAt10AM,
+    from,
+    to,
     tenantId: -1,
   });
 }

apps/stack/stacks/ExtractStack.ts

@@ -195,7 +195,7 @@ export function ExtractStack({ stack }: StackContext) {
 
   if (ENV.CRON_DISABLED !== 'true') {
     new Cron(stack, "ExtractCron", { 
-      schedule: "cron(00 10 * * ? *)",
+      schedule: "cron(8/15 * * * ? *)",
       job: {
         function: {
           handler: "src/extract/extract-tenants.cronHandler",

apps/stack/stacks/TransformStack.ts

@@ -70,7 +70,7 @@ export function TransformStack({ stack }: StackContext) {
 
   if (ENV.CRON_DISABLED !== 'true') {
     new Cron(stack, "TransformCron", {
-      schedule: "cron(00 13 * * ? *)",
+      schedule: "cron(0/15 * * * ? *)",
       job: {
         function: {
           handler: "src/transform/transform-tenant.cronHandler",