Self-hosting! (#49)

* added self-hosting

apps/worker/src/index.ts

@@ -5,6 +5,7 @@ import type { WorkerOptions } from 'bullmq';
 import { Worker } from 'bullmq';
 import express from 'express';
 
+import { createInitialSalts } from '@openpanel/db';
 import { cronQueue, eventsQueue, sessionsQueue } from '@openpanel/queue';
 import { getRedisQueue } from '@openpanel/redis';
 
@@ -15,7 +16,7 @@ import { register } from './metrics';
 
 const PORT = parseInt(process.env.WORKER_PORT || '3000', 10);
 const serverAdapter = new ExpressAdapter();
-serverAdapter.setBasePath(process.env.SELF_HOSTED ? '/worker' : '/');
+serverAdapter.setBasePath('/');
 const app = express();
 
 const workerOptions: WorkerOptions = {
@@ -162,10 +163,28 @@ async function start() {
     }
   );
 
+  if (process.env.SELF_HOSTED && process.env.NODE_ENV === 'production') {
+    await cronQueue.add(
+      'ping',
+      {
+        type: 'ping',
+        payload: undefined,
+      },
+      {
+        jobId: 'ping',
+        repeat: {
+          pattern: '0 0 * * *',
+        },
+      }
+    );
+  }
+
   const repeatableJobs = await cronQueue.getRepeatableJobs();
 
   console.log('Repeatable jobs:');
   console.log(repeatableJobs);
+
+  await createInitialSalts();
 }
 
 start();

apps/worker/src/jobs/cron.ping.ts

@@ -0,0 +1,26 @@
+import { chQuery, TABLE_NAMES } from '@openpanel/db';
+
+export async function ping() {
+  const [res] = await chQuery<{ count: number }>(
+    `SELECT COUNT(*) as count FROM ${TABLE_NAMES.events}`
+  );
+
+  if (typeof res?.count === 'number') {
+    const response = await fetch('https://api.openpanel.com/misc/ping', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({
+        domain: process.env.NEXT_PUBLIC_DASHBOARD_URL,
+        count: res?.count,
+      }),
+    });
+
+    if (response.ok) {
+      return await response.json();
+    }
+
+    throw new Error('Failed to ping the server');
+  }
+}

apps/worker/src/jobs/cron.ts

@@ -3,6 +3,7 @@ import type { Job } from 'bullmq';
 import { eventBuffer, profileBuffer } from '@openpanel/db';
 import type { CronQueuePayload } from '@openpanel/queue';
 
+import { ping } from './cron.ping';
 import { salt } from './cron.salt';
 
 export async function cronJob(job: Job<CronQueuePayload>) {
@@ -16,5 +17,8 @@ export async function cronJob(job: Job<CronQueuePayload>) {
     case 'flushProfiles': {
       return await profileBuffer.flush();
     }
+    case 'ping': {
+      return await ping();
+    }
   }
 }